Topic模式消息發(fā)送實(shí)例

1、pom引入

<dependency>            <groupId>junit</groupId>            <artifactId>junit</artifactId>            <version>4.11</version>            <scope>test</scope>        </dependency>        <dependency>            <groupId>org.apache.activemq</groupId>            <artifactId>activemq-all</artifactId>            <version>5.11.1</version>        </dependency>        <dependency>            <groupId>org.springframework</groupId>            <artifactId>spring-jms</artifactId>            <version>4.1.4.RELEASE</version>        </dependency>        <dependency>            <groupId>org.springframework</groupId>            <artifactId>spring-test</artifactId>            <version>4.1.4.RELEASE</version>        </dependency>

2、生產(chǎn)者配置

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"       xsi:schemaLocation="http://www.springframework.org/schema/beans        http://www.springframework.org/schema/beans/spring-beans.xsd">    <!-- 配置JMS連接工廠 -->    <bean id="providerConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">        <property name="brokerURL" value="failover:(tcp://192.168.147.131:61616)" />        <property name="useAsyncSend" value="true" />        <property name="clientID" value="providerClienctConnect" />    </bean>    <!-- 定義消息Destination -->    <bean id="topicDestination"  class="org.apache.activemq.command.ActiveMQTopic">        <constructor-arg value="testSpringTopic"/>    </bean>    <!-- 消息發(fā)送者客戶端 -->    <bean id="providerJmsTemplate" class="org.springframework.jms.core.JmsTemplate">        <property name="connectionFactory" ref="providerConnectionFactory" />        <property name="defaultDestination" ref="topicDestination" />        <!-- 開啟訂閱模式 -->        <property name="pubSubDomain" value="true"/>        <property name="receiveTimeout" value="10000" />        <!-- deliveryMode, priority, timeToLive 的開關(guān)要生效，必須配置為true，默認(rèn)false-->        <property name="explicitQosEnabled" value="true"/>        <!-- 發(fā)送模式             DeliveryMode.NON_PERSISTENT=1:非持久 ;             DeliveryMode.PERSISTENT=2:持久        -->        <property name="deliveryMode" value="1"/>    </bean>    </beans>

生產(chǎn)者程序

package com.mq.spring.topic;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.jms.core.JmsTemplate;import org.springframework.jms.core.MessageCreator;import org.springframework.test.context.ContextConfiguration;import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;import javax.annotation.Resource;import javax.jms.*;/** * created on 2015/6/4 * @author dennisit@163.com * @version 1.0 */@RunWith(SpringJUnit4ClassRunner.class)@ContextConfiguration(locations={"classpath:spring-topic.xml"})public class TopicSender {    @Resource(name = "providerJmsTemplate")    private JmsTemplate jmsTemplate;    @Test    public void send(){        sendMqMessage(null,"spring activemq topic type message[with listener] !");    }    /**     * 說明:發(fā)送的時(shí)候如果這里沒有顯示的指定destination.將用spring xml中配置的destination     * @param destination     * @param message     */    public void sendMqMessage(Destination destination, final String message){        if(null == destination){            destination = jmsTemplate.getDefaultDestination();        }        jmsTemplate.send(destination, new MessageCreator() {            public Message createMessage(session session) throws JMSException {                return session.createTextMessage(message);            }        });        System.out.println("spring send text message...");    }    public void setJmsTemplate(JmsTemplate jmsTemplate) {        this.jmsTemplate = jmsTemplate;    }}

3、消費(fèi)者配置

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"       xsi:schemaLocation="http://www.springframework.org/schema/beans        http://www.springframework.org/schema/beans/spring-beans.xsd">    <!-- 配置JMS連接工廠 -->    <bean id="consumerConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">        <property name="brokerURL" value="failover:(tcp://192.168.147.131:61616)" />        <property name="useAsyncSend" value="true" />        <property name="clientID" value="consumerClienctConnect" />    </bean>    <!-- 定義消息Destination -->    <bean id="topic1Destination"  class="org.apache.activemq.command.ActiveMQTopic">        <constructor-arg value="testSpringTopic1"/>    </bean>    <!-- 配置消息消費(fèi)監(jiān)聽者 -->    <bean id="consumerMessageListener" class="com.mq.spring.topic.ConsumerMessageListener" />    <!-- 消息訂閱客戶端1 -->    <bean id="consumerListenerClient1" class="org.springframework.jms.listener.DefaultMessageListenerContainer">        <property name="connectionFactory" ref="consumerConnectionFactory" />        <!-- 開啟訂閱模式 -->        <property name="pubSubDomain" value="true"/>        <property name="destination" ref="topic1Destination" />        <property name="subscr數(shù)據(jù)庫里，直到被這個(gè)ID的客戶端消費(fèi)掉-->        <property name="clientId" value="consumerClient1"/>        <property name="messageListener" ref="consumerMessageListener" />        <!-- 消息應(yīng)答方式             Session.AUTO_ACKNOWLEDGE  消息自動(dòng)簽收             Session.CLIENT_ACKNOWLEDGE  客戶端調(diào)用acknowledge方法手動(dòng)簽收             Session.DUPS_OK_ACKNOWLEDGE 不必必須簽收，消息可能會(huì)重復(fù)發(fā)送        -->        <property name="sessionAcknowledgeMode" value="1"/>    </bean>    <!-- 消息訂閱客戶端2 -->    <bean id="consumerListenerClient2" class="org.springframework.jms.listener.DefaultMessageListenerContainer">        <property name="connectionFactory" ref="consumerConnectionFactory" />        <!-- 開啟訂閱模式 -->        <property name="pubSubDomain" value="true"/>        <property name="destination" ref="topicDestination" />        <property name="subscriptionDurable" value="true"/>        <!---這里是設(shè)置接收客戶端的ID，在持久化時(shí)，但這個(gè)客戶端不在線時(shí)，消息就存在數(shù)據(jù)庫里，直到被這個(gè)ID的客戶端消費(fèi)掉-->        <property name="clientId" value="consumerClient2"/>        <property name="messageListener" ref="consumerMessageListener" />        <!-- 消息應(yīng)答方式             Session.AUTO_ACKNOWLEDGE  消息自動(dòng)簽收             Session.CLIENT_ACKNOWLEDGE  客戶端調(diào)用acknowledge方法手動(dòng)簽收             Session.DUPS_OK_ACKNOWLEDGE 不必必須簽收，消息可能會(huì)重復(fù)發(fā)送        -->        <property name="sessionAcknowledgeMode" value="1"/>    </bean></beans>

消費(fèi)者監(jiān)聽代碼

package com.mq.spring.topic;import org.apache.commons.lang.builder.ToStringBuilder;import javax.jms.JMSException;import javax.jms.Message;import javax.jms.MessageListener;import javax.jms.TextMessage;/** * created on 2015/6/4 * @author dennisit@163.com * @version 1.0 */public class ConsumerMessageListener implements MessageListener{    @Override    public void onMessage(Message message) {        TextMessage tm = (TextMessage) message;        try {            System.out.println("---------消息消費(fèi)---------");            System.out.println("消息內(nèi)容:/t" + tm.getText());            System.out.println("消息ID:/t" + tm.getJMSMessageID());            System.out.println("消息Destination:/t" + tm.getJMSDestination());            System.out.println("---------更多信息---------");            System.out.println(ToStringBuilder.reflectionToString(tm));            System.out.println("-------------------------");        } catch (JMSException e) {            e.printStackTrace();        }    }}

運(yùn)行結(jié)果:

說明:屬于學(xué)習(xí),網(wǎng)上資料結(jié)合個(gè)人理解,理解有誤的地方,期待指導(dǎo)和建議，共同學(xué)習(xí).

轉(zhuǎn)載請(qǐng)注明出處:[http://m.survivalescaperooms.com/dennisit/p/4552686.html]