Spring+Activemq(queue與topic模式)

一、queue與topic模式的區別

queue：一對一，一個消息只能由一個消費者接受

topic：一對多，廣播模式，需要消費者都在線（因此可能會導致消息的丟失）

二、pom依賴（使用jms連接active）

<properties>
      <springframework>4.3.14.RELEASE</springframework>
  </properties>
  
  <dependencies>
      <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-core</artifactId>
      <version>${springframework}</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-context</artifactId>
      <version>${springframework}</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-tx</artifactId>
      <version>${springframework}</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-webmvc</artifactId>
      <version>${springframework}</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-jms</artifactId>
      <version>${springframework}</version>
    </dependency>
    <!-- xbean 如<amq:connectionFactory /> -->
    <dependency>
      <groupId>org.apache.xbean</groupId>
      <artifactId>xbean-spring</artifactId>
      <version>3.16</version>
    </dependency>

    <!-- activemq -->
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-core</artifactId>
      <version>5.7.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-pool</artifactId>
      <version>5.12.1</version>
    </dependency>
  </dependencies>

三、spring-active.xml配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:amq="http://activemq.apache.org/schema/core"
       xmlns:jms="http://www.springframework.org/schema/jms"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:mvc="http://www.springframework.org/schema/mvc"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context-4.1.xsd
        http://www.springframework.org/schema/mvc
        http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd
        http://www.springframework.org/schema/jms
        http://www.springframework.org/schema/jms/spring-jms-4.1.xsd
        http://activemq.apache.org/schema/core
        http://activemq.apache.org/schema/core/activemq-core-5.12.1.xsd"
>
    <!-- 消息接收者掃描 -->
    <context:component-scan base-package="liuli.active.listener" ></context:component-scan>
    <!-- 運用xbean會導致經常連接不上，我也不知道什么問題 -->
<!--     <amq:connectionFactory id="amqConnectionFactory"
                           brokerURL="tcp://localhost:61616"
                           userName="admin"
                           password="admin" /> -->
                           
    <bean id="amqConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
          destroy-method="stop">
        <property name="connectionFactory">
            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                <property name="brokerURL">
                <!-- 注意端口號默認為61616 -->
                    <value>tcp://localhost:61616</value>
                </property>
                <property name="userName">
                    <value>admin</value>
                </property>
                <property name="password">
                    <value>admin</value>
                </property>
            </bean>
        </property>
        <property name="maxConnections" value="100"></property>
    </bean>

    <!-- 配置JMS連接工廠 -->
    <bean id="connectionFactory"
          class="org.springframework.jms.connection.CachingConnectionFactory">
        <constructor-arg ref="amqConnectionFactory" />
        <property name="sessionCacheSize" value="100" />
    </bean>

    <!-- 定義消息隊列（Queue） -->
    <bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <!-- 設置消息隊列的名字 -->
        <constructor-arg>
            <value>LiuliTest</value>
        </constructor-arg>
    </bean>
    <!-- 定義topic模式 -->
    <bean id="demoTopicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="liuliTopic"></constructor-arg>
    </bean>

    <!-- 配置JMS模板（Queue），Spring提供的JMS工具類，它發送、接收消息。 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="demoQueueDestination" />
        <property name="receiveTimeout" value="10000" />
        <!-- true是topic，false是queue，默認是false，此處顯示寫出false -->
        <property name="pubSubDomain" value="false" />
    </bean>
    
    <!-- 配置JMS模板（Topic） -->
    <bean id="jmsTemplateT" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="demoTopicDestination" />
        <property name="receiveTimeout" value="10000" />
        <!-- true是topic，false是queue，默認是false，此處顯示寫出false -->
        <property name="pubSubDomain" value="true" />
    </bean>

    <!-- 顯示注入消息監聽容器（Queue），配置連接工廠，監聽的目標是demoQueueDestination，監聽器是上面定義的監聽器  -->
    <bean id="queueListenerContainer"
          class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="demoQueueDestination" />
        <property name="messageListener" ref="queueMessageListener" />
    </bean>
    
    <!-- 顯示注入消息監聽容器（Topic），配置連接工廠，監聽的目標是demoQueueDestination，監聽器是上面定義的監聽器  -->
    <bean id="queueListenerContainer2"
          class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="demoTopicDestination" />
        <property name="messageListener" ref="queueMessageListenerT" />
    </bean>

</beans>

四、在這里我們使用springmvc模擬做消息發送

springmvc.xml配置略

五、寫消息發送接口代碼

package liuli.controller;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

@Controller
public class MyController {
    
    @Resource(name="jmsTemplateQ")
    private JmsTemplate jmsTemplate;
    @Resource(name="demoQueueDestination")
    private Destination destination;
    @Resource(name="jmsTemplateT")
    private JmsTemplate jmsTemplateT;
    @Resource(name="demoTopicDestination")
    private Destination destinationT;
    
    @RequestMapping(value="/send")
    @ResponseBody
    public String sendMsg(final String msg){
        jmsTemplate.send(destination, new MessageCreator() {
            @Override
            public Message createMessage(Session arg0) throws JMSException {
                return arg0.createTextMessage(msg);
            }
        });
        return msg;
    }
    
    @RequestMapping(value="/sendTopic")
    @ResponseBody
    public String sendTopicMsg(final String msg){
        jmsTemplateT.send(destinationT, new MessageCreator() {
            @Override
            public Message createMessage(Session arg0) throws JMSException {
                return arg0.createTextMessage(msg);
            }
        });
        return msg;
    }

}

六、寫消息監聽代碼（當然，正常情況下消息發送與接受應該在不同工程）

package liuli.active.listener;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.springframework.stereotype.Component;

@Component
public class QueueMessageListener implements MessageListener{

    @Override
    public void onMessage(Message message) {
        TextMessage ms = (TextMessage)message;
        try {
            System.out.println("queue---消息----"+ms.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

package liuli.active.listener;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.springframework.stereotype.Component;

@Component
public class QueueMessageListenerT implements MessageListener{

    @Override
    public void onMessage(Message message) {
        TextMessage ms = (TextMessage)message;
        try {
            System.out.println("topic---消息----"+ms.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

七、配置web.xml（略）

八、下載activemq並啟動（略）

九、啟動tomcat即可