ActiveMQ

2019-11-14 22:12:33

字體：大中小

來源：轉載

供稿：網友

ActiveMQ - 入門指南

首先需要下載ActiveMQ，下面的鏈接給我們列出了所有版本：http://activemq.apache.org/download-archives.html每個版本為不同的OS提供了鏈接：

公司電腦是windows的，用目錄下的activemq.bat啟動：

端口號默認是61616，可以在conf/activemq.xml中看到:

<transportConnectors>    <!-- DOS PRotection, limit concurrent connections to 1000 and frame size to 100MB -->    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>    <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>    <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>    <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>    <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/></transportConnectors>

相關的Maven dependency：

<dependency>    <groupId>javax.jms</groupId>    <artifactId>javax.jms-api</artifactId>    <version>2.0</version></dependency><dependency>    <groupId>org.apache.activemq</groupId>    <artifactId>activemq-core</artifactId>    <version>5.7.0</version></dependency>

使用javax.jms.session跟JMS Provider通信，好像說了句廢話...：

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(        ActiveMQConnection.DEFAULT_USER,        ActiveMQConnection.DEFAULT_PASSWord, "tcp://localhost:61616");Connection connection = connectionFactory.createConnection();connection.start();Session session = connection.createSession(Boolean.TRUE,        Session.AUTO_ACKNOWLEDGE);

然后一些目的地、發送者、發送內容什么的都是由session來弄的：

Destination destination = session.createQueue("this is sparta!!");MessageProducer producer = session.createProducer(destination);TextMessage message0 = session.createTextMessage("這是斯巴達!!!");TextMessage message1 = session.createTextMessage("這也是斯巴達!!!");TextMessage message2 = session.createTextMessage("這些都是斯巴達!!!");producer.send(message0);producer.send(message1);producer.send(message2);session.commit();

有了producer，相應地也有consumer，接收消息方法如下：

MessageConsumer consumer = session.createConsumer(destination);System.out.println(((TextMessage) consumer.receive(10000)).getText());

結果還是consumer去一個個receive了，就像是接收人親自去確認那樣。或許我們可以讓Listener代勞:

consumer.setMessageListener(new MessageListener() {    public void onMessage(Message message) {        try {            System.out.println("listener catched:::"+((TextMessage)message).getText());        } catch (JMSException e) {            e.printStackTrace();        }    }});

當這個consumer設置了Listener的時候就不能再以receive()的方式接收了,不然會出現javax.jms.IllegalStateException:Cannot synchronously receive a message when a MessageListener is set...

如果想使用publish/subscribe，直接將createQueue改為createTopic即可，但需要理解Topic是無狀態的。

完整code如下，發送者：

{    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(            ActiveMQConnection.DEFAULT_USER,            ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");    Connection connection = connectionFactory.createConnection();    connection.start();    Session session = connection.createSession(Boolean.TRUE,            Session.AUTO_ACKNOWLEDGE);    Destination destination = session.createQueue("this is sparta!!");    MessageProducer producer = session.createProducer(destination);    TextMessage message0 = session.createTextMessage("這是斯巴達!!!");    TextMessage message1 = session.createTextMessage("這也是斯巴達!!!");    TextMessage message2 = session.createTextMessage("這些都是斯巴達!!!");    producer.send(message0);    producer.send(message1);    producer.send(message2);    session.commit();}

接收者：

{    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(            ActiveMQConnection.DEFAULT_USER,            ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");;    Connection connection = connectionFactory.createConnection();    connection.start();    Session session = connection.createSession(Boolean.FALSE,            Session.AUTO_ACKNOWLEDGE);    Destination destination = session.createQueue("this is sparta!!");    MessageConsumer consumer = session.createConsumer(destination);    System.out.println(((TextMessage) consumer.receive(10000)).getText());    System.out.println(((TextMessage) consumer.receive(10000)).getText());    System.out.println(((TextMessage) consumer.receive(10000)).getText());}

這次試試集成到Spring。下面的連接是ActiveMQ官網提供的文檔。http://activemq.apache.org/spring-support.html

下面是我添加的一些dependency，基本的spring依賴我就不列舉了:

<!-- jms activemq --><dependency>    <groupId>javax.jms</groupId>    <artifactId>javax.jms-api</artifactId>    <version>2.0</version></dependency><dependency>    <groupId>org.apache.activemq</groupId>    <artifactId>activemq-core</artifactId>    <version>${activemq.version}</version></dependency><dependency>    <groupId>org.apache.activemq</groupId>    <artifactId>activemq-pool</artifactId>    <version>${activemq.version}</version></dependency><dependency>    <groupId>org.apache.activemq</groupId>    <artifactId>activemq-spring</artifactId>    <version>${activemq.version}</version></dependency><dependency>    <groupId>org.apache.xbean</groupId>    <artifactId>xbean-spring</artifactId>    <version>3.16</version></dependency><dependency><dependency>    <groupId>org.springframework</groupId>    <artifactId>spring-jms</artifactId>    <version>${spring.version}</version></dependency>

maven中添加時要注意還有個xbean-spring；之前并沒有注意，運行發現異常提示 ClassNotFound:org.apache.xbean.spring.context.v2.XBeanNamespaceHandler；

后來我添加了xbean-v2，結果提示v2c，于是我添加v2c，后來感覺太傻就加了xbean-spring。配置方面可以使用jms和activeMq的標簽:

xmlns:amq="http://activemq.apache.org/schema/core"xmlns:jms="http://www.springframework.org/schema/jms"

相應的xsi:schemaLocation:

http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsdhttp://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms.xsd

關于connectionFactory的配置可以使用amq標簽:

<amq:connectionFactory id="jmsFactory" brokerURL="tcp://localhost:61616" />

但是在這里我打算試試PooledConnectionFactory；關于org.apache.activemq.pool.PooledConnectionFactory官網有以下解釋(簡單到位，都不用翻譯了):

If you are not using a JCA container to manage your JMS connections, we recommend you use our pooling JMS connection provider, (org.apache.activemq.pool.PooledConnectionFactory) from the activemq-pool library, which will pool the JMS resources to work efficiently with Spring's JmsTemplate or with EJBs.

對于其屬性，下面根據javaDoc給出一些解釋：

MaximumActiveSessionPerConnection：每個Connection的最大Session數
BlockIfSessionPoolIsFull：默認為session池滿時請求獲得session會阻塞；設置false則會拋出JMSException
MaxConnections：最大連接數
IdleTimeout：空閑時間，默認為30秒
CreateConnectionOnStartup：是否在啟動時創建connection

在這里我先用默認參數聲明，不知道為什么總是報MalformPrameterizedType...

<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" />

上次用的隊列，這次換用Topic試試...

<bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">  <constructor-arg index="0" value="spartaTopic"></constructor-arg></bean>

當然也可以使用amq標簽:

<amq:topic physicalName="sparta" />

如果是使用queue的話:

<amq:queue physicalName="sparta" />

難道我將這些放到spring里就是為了用用標簽方便DI?用里的話來說就是jmsTemplate是<spring對jms支持核心的部分>(另有jmsTemplate102為適應jms1.0.2的)；和jdbcTemplate那樣 jmsTemplate也有提供相似的優勢。比如，像jdbcTemplate處理失控的jdbc代碼那樣，用jmsTemplate處理失控的jms代碼。或者，如果在使用JmsTemplate是捕捉到了JMSException，JmsTemplate將捕獲該異常，然后拋出一個Spring自帶的JmsException的子類異常(個人感覺spring提供的不是更詳細的異常信息，只是側重點不同...)。

比如：

ListenerExecutionFailedException：監聽器執行失敗
MessageConversionException：消息轉換失敗
SynchedLocalTransactionFailedException：同步本地事務未完成
UncategorizedJmsException：沒有適合異常的其他情況

如果我們catch了JMSException，我們依然可以把他轉為JmsException:

catch (JMSException e) {    e.printStackTrace();    JmsException je = JmsUtils.convertJmsaccessException(e);}

現在試著配置jmsTemplate：

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate" >    <property name="connectionFactory" >        <bean class="org.apache.activemq.pool.PooledConnectionFactory" />    </property>    <property name="defaultDestination" >        <amq:topic physicalName="sparta" />    </property></bean>

這樣編寫代碼時就變得簡單多了，之前那些connectionFactory,connection,session,consumer,producer統統不見了；我只需要(匿名內部類可能有些礙眼):

applicationContext context = new ClassPathXmlApplicationContext("classpath:applicationContext.xml");JmsTemplate template = (JmsTemplate)context.getBean("jmsTemplate");template.send(new MessageCreator() {    public Message createMessage(Session session) throws JMSException {        ActiveMQMapMessage msg = (ActiveMQMapMessage)session.createMapMessage();        msg.setString("msg", "This is sparta!!");        return msg;    }});

接收時只需要：

template.receive();

但需要注意！這個receive是同步接收消息的，他會一直阻塞到有消息個接收。可能會想到MessageListener，比如我們可以給一個MessageConsumer對象setMessageListener:

MessageConsumer consumer = session.createConsumer(destination);consumer.setMessageListener(new MessageListener() {    public void onMessage(Message message) {        try {            System.out.println("listener catched:::"+((TextMessage)message).getText());        } catch (JMSException e) {            e.printStackTrace();        }    }});

上面代碼中的MessageListener實例，如果新建一個專門用來監聽的類，實現MessageListener接口并加上MessageDriven標注就會出現一個問題——他不夠pojo。他有侵入性，我不想要任何實現接口的語法出現在代碼中。

于是我可以用listener-container；現在我創建一個類去監聽，比如:

public class CustomedListener {    void processHandle(HashMap<String,String> map){        System.out.println("msg:::"+map.get("msg"));    }}

但需要注意的是方法的參數列表，他并不是隨便定義的。上面的publisher發送的message是ActiveMQMapMessage，這就需要我把參數定義為上面那種形式。然后看一下spring中如何配置這個Listener:

<bean id="myListener" class="pac.testcase.jms.CustomedListener"/><jms:listener-container connection-factory="connectionFactory">    <jms:listener destination="sparta" ref="myListener" method="processHandle"/></jms:listener-container>

這樣我就不需要去調用receive了，有消息就接收。

現在試試通過JMS，在應用程序之間發送消息。先看看spring提供的RPC方案(其實還有其他方案，只是沒見過誰用)。需要使用到這兩個類:

org.springframework.jms.remoting.JmsInvokerServiceExporter將bean導出為基于消息的服務
org.springframework.jms.remoting.JmsInvokerProxyFactoryBean讓客戶端調用服務

比較一下JmsInvokerServiceExporter和RmiServiceExporter:

我創建一個接口和實現類如下:

package pac.testcase.jms;public interface JmsRmiService {    String doServe(String requestedNum);}

實現：

package pac.testcase.jms;import org.springframework.stereotype.Service;@Servicepublic class JmsRmiServiceImpl implements JmsRmiService {    public String doServe(String content) {        System.out.println(content.concat(" has been requested!!"));        return "your message::".concat(content).concat(":::length:")+content.length();    }}

將這個pojo聲明為服務，在spring配置文件中配置:

<bean id="serverService" class="org.springframework.jms.remoting.JmsInvokerServiceExporter"    p:serviceInterface="pac.testcase.jms.JmsRmiService"    p:service-ref="JmsRmiServiceImpl"></bean>

需將他設置為jms監聽器，配置方法和一般的jmsMessageListener的配置相同：

<amq:connectionFactory id="jmsFactory" /><jms:listener-container    destination-type="queue"    connection-factory="jmsFactory"    concurrency="3"    container-type="simple">    <jms:listener  destination="sparta" ref="serverService"  /></jms:listener-container>

container-type有simple和default，根據不同的type也可以使用task-Executor，這里先簡單記錄一下。先啟動jms broker再啟動:

new ClassPathXmlApplicationContext("classpath:applicationContext-*.xml").getBean(JmsRmiService.class);

client這邊我需要一個調用代理幫我去調用接口，也就是JmsInvokerProxyFactoryBean；配置如下:

<amq:connectionFactory id="connectionFactory" /><bean id="clientService" class="org.springframework.jms.remoting.JmsInvokerProxyFactoryBean"    p:serviceInterface="pac.test.jms.SenderRmiService"    p:connectionFactory-ref="connectionFactory"    p:queueName="sparta"/>

配置中的serviceInterface是client端中根據要調用的方法創建的一個接口。main方法試著調用看看:

public static void main(String[] args) {    ApplicationContext context = new ClassPathXmlApplicationContext("classpath:applicationContext.xml");    SenderRmiService service = (SenderRmiService)context.getBean("clientService");    System.out.println(service.doServe("這才是斯巴達!!"));}

server端輸出：