ESB內部所交流/傳遞的都是消息,所以可見(jiàn)這個(gè)消息格式的重要性. 在JBossESB中, 定義了一個(gè)Message的對象,它是有以下幾個(gè)部分構成的.

    
Header (用來(lái)存放From, To, Reply-to等Addressing的信息).    
Body (存放信息主體)    
Attachment （用來(lái)存放附件等)    
Properties    
Context (主要是存放一些類(lèi)似事務(wù)的信息等)

目前在Body里面一般來(lái)說(shuō)存放兩種格式的數據,一個(gè)是串行化數據(Serialized Object ),另外一個(gè)是XML文件,比如常見(jiàn) 的SOAP的payload. 在ESB中,還有兩個(gè)定義,一個(gè)叫ESB-aware Message, 我們上面所講的Message就是ESB-aware Message, 正如名字說(shuō)講的,它是屬于ESB內部的Message對象. 還有個(gè)叫 ESB unaware Message,也就是說(shuō)他同樣也是一個(gè)message,比如SOAP Message,但是如果把soap message直接讓ESB來(lái)處理，是處理不了的,所以呢? 經(jīng)常的在Listener 監聽(tīng)的端口會(huì )有個(gè)Adapter (在JBossESB里叫做Gateway)來(lái)負責把ESB-unaware message 轉成 ESB-aware message.

ESB的內部服務(wù)是用EPR來(lái)映射的. ESB的內部服務(wù)可以是任何的一個(gè)服務(wù),比如說(shuō)一個(gè)FTP的服務(wù),一個(gè)基于文件系統的服務(wù)等等, 那么這個(gè)時(shí)候我們就需要用EPR來(lái)對這個(gè)服務(wù)進(jìn)行描述.在EPR這個(gè)類(lèi)里,主要是描述了這個(gè)服務(wù)的URI,以及所必須的一些元數據. 目前在JBossESB中提供的EPR有: FileEPR,EmailEPR,FTPEPR, HibernateEPR等等. 我們在注冊服務(wù)的時(shí)候,是將EPR的信息注冊到UDDI的容器里, 但不僅僅是EPR, 還有一些輔助信息,比如定義服務(wù)的category-name, service-name. 這些將在后面繼續介紹.

Listener的作用是負責監聽(tīng)端口,一般來(lái)說(shuō),客戶(hù)端是發(fā)送消息到Listener,然后有Listener把消息傳遞給ESB, 我們可以把Listener看做是inbound router. 在JBossESB中,我們是叫GatewayListener, 它一般來(lái)說(shuō)做兩件事情.

目前ESB支持的Gateway有: JMSGatewayListener, JBossRemotingGatewayListener, FileGatewayListener等等. 在MessageComposer這個(gè)類(lèi)里的compose/decompose方法來(lái)負責ESB-unaware信息和ESB-aware信息的轉化.

public interface MessageComposer<T> {
    /**
     * Set the composer's configuration
     */
    public void setConfiguration(ConfigTree config) throws ConfigurationException;
    /**
     * Compose an ESB "aware" message from the supplied message payload.
     * Implementations need to construct and populate an ESB Message from the
     * messagePayload instance.
     */
    public Message compose(T messagePayload) throws MessageDeliverException;
    /**
     * Decompose an ESB "aware" message, extracting and returning the message payload.
     */
    public Object decompose(Message message, T originalInputMessagePayload) throws MessageDeliverException;
}

Courier的作用就是負責傳輸,正如以下接口所顯示:

public interface Courier extends DeliverOnlyCourier{public boolean deliver(Message message) 	throwsCourierException,MalformedEPRException;}

目前實(shí)現的Transport有:JmsCourier, InVMCourier, HibernateCourier, FileCourier等傳輸層,在ESB內部是通過(guò)EPR來(lái)跟Courier進(jìn)行關(guān)聯(lián)的.

在JBossESB中,我們可以把ActionProcessingPipeline類(lèi)看作是Message Filter, 每個(gè)Message都會(huì )經(jīng)過(guò) ActionProcessingPipeline的處理. 里面有這么個(gè)方法:

public boolean process(final Message message)

而actionProcessingPipeline又是由Action (ActionPipelineProcessor)來(lái)組成的. 我們可以把Action看成是Interceptor, 由它來(lái)實(shí)現具體的業(yè)務(wù)邏輯,可以是路由,又或者數據轉化功能等等. 如果你用JBossESB的話(huà),那么Action是一個(gè)非常重要的部分,我們來(lái)看下它所定義的接口.

public interface ActionPipelineProcessor extends ActionLifecycle{public Message process(final Message message) throws ActionProcessingException ;public void processException(final Message message, final Throwable th) ;public void processSuccess(final Message message) ;}

一般來(lái)說(shuō)自定義的Action把具體的業(yè)務(wù)邏輯放在Process的方法里,當然了,你也可以定義相對應的Exception處理方法,通過(guò)實(shí)現processException.在ESB代碼中,自定義的Action可以用繼承AbstractActionPipelineProcessor 或者 AbstractActionLifecycle.

在有些情況下,你需要一些全局的Interceptor,我們之前說(shuō)的Action,可以理解成是每個(gè)service的interceptor,但是如果我需要使用log來(lái)記錄一個(gè)消息在各個(gè)service之間傳輸的日志, 又或者想記錄消息進(jìn)入某個(gè)service的時(shí)間和退出的時(shí)間. 那么在JBoss ESB中就有Filter的概念. 如果你要實(shí)現自己的Filter,需要繼承InputOputFilter類(lèi).

public class InputOutputFilter
{   
    /**
     * Called as the message flows towards the transport.
     */
    
    public Message onOutput (Message msg, Map<String, Object> params) throws CourierException
    {
        return msg;
    }
    
    /**
     * Called immediately after the message is received from the transport.
     */
    
    public Message onInput (Message msg, Map<String, Object> params) throws CourierException
    {
        return msg;
    }
}

onInput方法總是在從傳輸層獲取到Message后,第一步所做的工作;類(lèi)似的, onOutput是給傳輸層傳遞前所做的最后一步工作. 你可以在TwoWayCourierImpl中看到這段代碼的調用.

對于客戶(hù)端調用來(lái)說(shuō),EPR, Courier等都太底層了.所以如果對此進(jìn)行了封裝. 我們對每個(gè)service加以service-category和service-name的屬性. 所以如果你想發(fā)送一個(gè)ESB的內部Message,你只需要知道目標service的service-category和service-name,然后就可以調用ServiceInvoker來(lái)調用服務(wù). 不需要去使用Courier等底層的API, 另外用ServiceInvoker還可以支持fail-over等特性.

public class ServiceInvoker {
        
        public ServiceInvoker(String serviceCategory, String serviceName) throws MessageDeliverException {
                    this(new Service(serviceCategory, serviceName));
        }
            
        public Message deliverSync(Message message, long timeoutMillis) throws MessageDeliverException, RegistryException, FaultMessageException
        public void deliverAsync(Message message) throws MessageDeliverException
}
        

下面我們來(lái)具體看下helloworld_action這個(gè)例子. 在JBoss ESB 4.4的例子中,我們默認以JBoss Messaging來(lái)作為JMS的實(shí)現.

    <providers>
          <jms-provider name="JBossMQ" connection-factory="ConnectionFactory">
                      
              <jms-bus busid="quickstartGwChannel">
                  <jms-message-filter
                      dest-type="QUEUE"
                      dest-name="queue/quickstart_helloworld_action_Request"
                  />
              </jms-bus>
              <jms-bus busid="quickstartEsbChannel">
                  <jms-message-filter
                      dest-type="QUEUE"
                      dest-name="queue/B"
                  />
              </jms-bus>

          </jms-provider>
      </providers>
               

<services>
          
        <service category="HelloWorld_ActionESB" 
                 name="SimpleListener" 
                 description="Hello World" >
            <listeners>
                <jms-listener name="JMS-Gateway"
                    busidref="quickstartGwChannel"                         
                    is-gateway="true"
                />
                <jms-listener name="JMS-ESBListener"
                              busidref="quickstartEsbChannel"
                />                
            </listeners>
            <actions mep="OneWay">
            <action name="action2"
                    class="org.jboss.soa.esb.actions.SystemPrintln"
                    />
               <action name="displayAction" 
                    class="org.jboss.soa.esb.samples.quickstart.helloworldaction.MyJMSListenerAction" 
                    process="displayMessage">
                    <property name="exceptionMethod" value="exceptionHandler"/>
               </action>
               <action name="playAction" 
                    class="org.jboss.soa.esb.samples.quickstart.helloworldaction.MyJMSListenerAction" 
                    process="playWithMessage">        
                    <property name="exceptionMethod" value="exceptionHandler"/>
               </action>  
            </actions>
        </service>
      </services>
          

public class MyJMSListenerAction extends AbstractActionLifecycle
{
   protected ConfigTree _config;
          
   public MyJMSListenerAction(ConfigTree config) { _config = config; }  
          
    public Message playWithMessage(Message message) throws Exception {
       Body msgBody = message.getBody();
       String contents = msgBody.get().toString();
       StringBuffer sb = new StringBuffer();
       sb.append("\nBEFORE**\n");
       sb.append(contents);
       sb.append("\nAFTER**\n");
       msgBody.add(sb.toString());
       return message;
   }
}
           

    
/META-INF/jboss-esb.xml    
 /META-INF/deployment.xml 在這里定義對其他包或者服務(wù)的依賴(lài),或者配置classloader.    
jbm-queue-service.xml (optional) 這里是定義啟動(dòng)所需要的Queue    
lib (optional) 放些所需要的第三方包    
所需要的些classes文件

目前在JBossESB中,一般有兩種方式來(lái)調用service. 一種是通過(guò)Gateway listener, 另外一種是直接通過(guò)ServiceInvoker的API來(lái)調用.

public class SendJMSMessage {
              
  public void setupConnection() throws JMSException, NamingException
    {
        InitialContext iniCtx = new InitialContext();
        Object tmp = iniCtx.lookup("ConnectionFactory");
        QueueConnectionFactory qcf = (QueueConnectionFactory) tmp;
        conn = qcf.createQueueConnection();
        que = (Queue) iniCtx.lookup("queue/quickstart_helloworld_action_Request");
        session = conn.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
        conn.start();
        System.out.println("Connection Started");
    }
              
              
    public void sendAMessage(String msg) throws JMSException {  
        QueueSender send = session.createSender(que);        
        ObjectMessage tm = session.createObjectMessage(msg);
         tm.setStringProperty(StoreMessageToFile.PROPERTY_JBESB_FILENAME, "HelloWorldActionTest.log");
        send.send(tm);        
        send.close();
    }
       
    
    public static void main(String args[]) throws Exception
    {                   
        SendJMSMessage sm = new SendJMSMessage();
        sm.setupConnection();
        sm.sendAMessage(args[0]); 
        sm.stop();
    }
}
          

public class SendEsbMessage 
{
    public static void main(String args[]) throws Exception
    {
//      Setting the ConnectionFactory such that it will use scout
        System.setProperty("javax.xml.registry.ConnectionFactoryClass","org.apache.ws.scout.registry.ConnectionFactoryImpl");
        
        if (args.length < 3)
        {
            System.out.println("Usage SendEsbMessage <category> <name> <text to send>");
        }
        
        Message esbMessage = MessageFactory.getInstance().getMessage();
        esbMessage.getBody().add(args[2]);  
        new ServiceInvoker(args[0], args[1]).deliverAsync(esbMessage);
        
    }   
}

我們結合之前的概念,來(lái)看下這個(gè)例子的調用過(guò)程. 這里我們假設是通過(guò)JMS Gateway來(lái)調用ESB服務(wù)的.

我們這里講述了一個(gè)簡(jiǎn)單的調用oneway服務(wù)的一個(gè)過(guò)程.

因為JBoss ESB 4.系列是基于JBoss AS的,所以有個(gè)擴展類(lèi)JBoss4ESBDeployer類(lèi)是來(lái)負責對JBoss ESB包的監聽(tīng)和解析. 我們可以看到在$jbossesb.sar/META-INF/jboss-service.xml里面定義了JBoss4ESBDeployer的服務(wù).

 <mbean code="org.jboss.soa.esb.listeners.config.JBoss4ESBDeployer"
       name="jboss.esb:service=ESBDeployer">
  <depends>jboss.esb:service=ESBRegistry</depends>
  <depends>jboss.esb:service=JuddiRMI</depends>
</mbean>

所以,通過(guò)部署了JBoss4ESBDeployer,凡是以.esb結尾的包,都是由JBoss4ESBDeployer來(lái)解析, 如果你對這塊比較感興趣,還可以看下以下的幾個(gè)類(lèi).

		   org.jboss.soa.esb.listeners.config.ConfigurationControllerorg.jboss.soa.esb.listeners.config.JBoss4ESBDeploymentorg.jboss.soa.esb.listeners.config.Generator

正如我們之前所看到的,因為JBossESB內部本身的架構比較靈活,所以說(shuō)你只要寫(xiě)個(gè)自己的Action,可以很方面的做和其他library的集成. 直到JBoss ESB 4.4.GA版本,有以下一系列的集成.