使用activemq

说说在项目里是怎么使用activemq(简称为amq)进行通信的。

有2个系统,面向不同的用户,简称为系统A和系统B。本文为了简单,只将系统A作为 队列A.CreateMessage的生产端,系统B作为 队列A.CreateMessage的消费端,传输的message可为一个设计好的类的对象,本文为了简单,传输的是一个String对象。

另外,系统A也可以作为另一队列QC的消费端,系统B作为队列QC的生产端。

 1.下载一个apache-activemq-5.10.2,根据系统类型(操作系统位数),选择启动bin目录下的win32或win64目录下的activemq.bat文件。启动后,打开浏览器,输入localhost:8161/admin/queues.jsp,

如果页面是下面这样的

      使用activemq

输入用户名:admin,密码:admin就OK了。

           使用activemq

2.amq也启动了,那么接下来是在系统A加上amq相关内容。

项目目录结构如下:

使用activemq

系统A的applicationContext-amq.xml文件:

使用activemq
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:amq="http://activemq.apache.org/schema/core"
    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
 http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.10.2.xsd">
    <!--
        使用spring的listenerContainer,消息用持久化保存,服务器重启不会丢失
    -->
     <!-- 连接外部的activeMQ-->
    <amq:connectionFactory id="jmsConnectionFactory" userName="admin" password="admin" brokerURL="tcp://localhost:61616"></amq:connectionFactory>

    <!--  Spring JmsTemplate config -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory">
            <!--  lets wrap in a pool to avoid creating a connection per send -->
            <bean class="org.springframework.jms.connection.SingleConnectionFactory">
                <property name="targetConnectionFactory" ref="jmsConnectionFactory" />
            </bean>
        </property>
        <!-- custom MessageConverter -->
        <property name="messageConverter" ref="defaultMessageConverter" />
    </bean>
    
    <!-- converter  -->
    <bean id="defaultMessageConverter" class="com.pack.app.amq.DefaultMessageConverter" />
    
    <!--  ActiveMQ destinations  -->
    <!--  使用Queue方式-->
    <amq:queue name="QUEUE" physicalName="TESTQ" />

    <bean id="queueMessageProducer" class="tools.amq.QueueMessageProducer">
        <property name="template" ref="jmsTemplate" />
        <property name="destination" ref="QUEUE" />
    </bean>
    
  
使用activemq
  <!-- consumer for queue -->
    <bean id="queueConsumer" class="tools.amq.QueueConsumer" />

    <!-- Message Listener for queue -->
    <bean id="queueListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
        <constructor-arg ref="queueConsumer" />
        <!--  may be other method -->
        <property name="defaultListenerMethod" value="receive" />
        <!-- custom MessageConverter define -->
        <property name="messageConverter" ref="defaultMessageConverter" />
    </bean>

    <!--  listener container,MDP无需实现接口 -->
    <bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="jmsConnectionFactory" />
        <property name="destination" ref="QUEUE" />
        <property name="messageListener" ref="queueListener" />
    </bean>
使用activemq
<!-- 测试 向MQ发消息 -->
    <amq:queue name="CreateMessage" physicalName="A.CreateMessage" />
    <!-- 生产数据 -->
    <bean id="createMessageProducer" class="com.pack.app.amq.producer.CreateMessageProducer">
        <property name="template" ref="jmsTemplate" />
        <property name="destination" ref="CreateMessage" />
    </bean>
</beans>
使用activemq

DefaultMessageConverter.java

使用activemq
public class DefaultMessageConverter implements MessageConverter {
    /**
     * Logger for this class
     */
    private static final Log log = LogFactory.getLog(DefaultMessageConverter.class);

    public Message toMessage(Object obj, Session session) throws JMSException {
        if (log.isDebugEnabled()) {
            log.debug("toMessage(Object, Session) - start");
        }

        // check Type
        ActiveMQObjectMessage objMsg = (ActiveMQObjectMessage) session.createObjectMessage();
        HashMap<String, byte[]> map = new HashMap<String, byte[]>();
        try {
            // POJO must implements Seralizable
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(bos);
            oos.writeObject(obj);
            map.put("POJO", bos.toByteArray());
            objMsg.setObjectProperty("Map", map);

        } catch (IOException e) {
            e.printStackTrace();
            log.error("toMessage(Object, Session)", e);
        }
        return objMsg;
    }

    public Object fromMessage(Message msg) throws JMSException {
        if (log.isDebugEnabled()) {
            log.debug("fromMessage(Message) - start");
        }

        if (msg instanceof ObjectMessage) {
            HashMap<String, byte[]> map = (HashMap<String, byte[]>) ((ObjectMessage) msg).getObjectProperty("Map");
            try {
                // POJO must implements Seralizable
                ByteArrayInputStream bis = new ByteArrayInputStream(map.get("POJO"));
                ObjectInputStream ois = new ObjectInputStream(bis);
                Object returnObject = ois.readObject();
                return returnObject;
            } catch (IOException e) {
                e.printStackTrace();
                log.error("fromMessage(Message)", e);

            } catch (ClassNotFoundException e) {
                e.printStackTrace();
                log.error("fromMessage(Message)", e);
            }

            return null;
        } else {
            throw new JMSException("Msg:[" + msg + "] is not Map");
        }
    }
}
使用activemq
QueueMessageProducer.java
使用activemq
import org.springframework.jms.core.JmsTemplate;

import javax.jms.Queue;

/**
 * Date: 2015-7-1
 * Time: 17:14:23
 */
public class QueueMessageProducer {

    private JmsTemplate template;

    private Queue destination;

    public void setTemplate(JmsTemplate template) {
        this.template = template;
    }

    public void setDestination(Queue destination) {
        this.destination = destination;
    }

    public void send(FooMessage message) {
        template.convertAndSend(this.destination, message);
    }

}
使用activemq

CreateMessageProducer.java(消息生产者)

使用activemq
import javax.jms.Queue;
import org.springframework.jms.core.JmsTemplate;

public class CreateMessageProducer {

     private JmsTemplate template;

     private Queue destination;
     
     public void setTemplate(JmsTemplate template) {
        this.template = template;
    }

    public void setDestination(Queue destination) {
        this.destination = destination;
    }

    public void send(String str) {
        template.convertAndSend(this.destination, str);
        System.out.println("system A send message to system B~~~~~~~~~~");
    }
}
使用activemq

 3.在系统B加上amq相关内容。

applicationContext-amq.xml文件
使用activemq
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:amq="http://activemq.apache.org/schema/core"
    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
 http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.10.2.xsd">
    <!--
        使用spring的listenerContainer,消息用持久化保存,服务器重启不会丢失
    -->
     <!-- 连接外部的activeMQ-->
    <amq:connectionFactory id="jmsConnectionFactory" userName="admin" password="admin" brokerURL="tcp://localhost:61616"></amq:connectionFactory>

    <!--  ActiveMQ destinations  -->
    <!--  使用Queue方式-->
    <amq:queue name="QUEUE" physicalName="TESTQ" />

    <!--  Spring JmsTemplate config -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory">
            <!--  lets wrap in a pool to avoid creating a connection per send -->
            <bean class="org.springframework.jms.connection.SingleConnectionFactory">
                <property name="targetConnectionFactory" ref="jmsConnectionFactory" />
            </bean>
        </property>
        <!-- custom MessageConverter -->
        <property name="messageConverter" ref="defaultMessageConverter" />
    </bean>

    <!-- converter  -->
    <bean id="defaultMessageConverter" class="com.pack.app.amq.DefaultMessageConverter" />

    <bean id="queueMessageProducer" class="tools.amq.QueueMessageProducer">
        <property name="template" ref="jmsTemplate" />
        <property name="destination" ref="QUEUE" />
    </bean>

    <!-- consumer for queue -->
    <bean id="queueConsumer" class="tools.amq.QueueConsumer" />

    <!-- Message Listener for queue -->
    <bean id="queueListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
        <constructor-arg ref="queueConsumer" />
        <!--  may be other method -->
        <property name="defaultListenerMethod" value="receive" />
        <!-- custom MessageConverter define -->
        <property name="messageConverter" ref="defaultMessageConverter" />
    </bean>

    <!--  listener container,MDP无需实现接口 -->
    <bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="jmsConnectionFactory" />
        <property name="destination" ref="QUEUE" />
        <property name="messageListener" ref="queueListener" />
    </bean>
    
    
    <!-- 测试  接收消息 -->
    <amq:queue name="CreateMessage" physicalName="A.CreateMessage" />
    <!-- 接收数据 -->
    <bean id="createMessageConsumer" class="com.pack.app.amq.consumer.CreateMessageConsumer" />  
    <!-- 监听 -->
    <bean id="createMessageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
        <constructor-arg ref="createMessageConsumer" />
        <!--  may be other method -->
        <property name="defaultListenerMethod" value="process" />
        <!-- custom MessageConverter define -->
        <property name="messageConverter" ref="defaultMessageConverter" />
    </bean>

    <!--  listener container,MDP无需实现接口 -->
    <bean id="createMessageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="jmsConnectionFactory" />
        <property name="destination" ref="CreateMessage" />
        <property name="messageListener" ref="createMessageListener" />
        
        <!-- 消费者个数  -->
        <!-- <property name="concurrentConsumers" value="4"></property> -->
    </bean>
</beans>
使用activemq
DefaultMessageConverter.java、QueueMessageProducer.java、QueueConsumer.java与系统A一样。

CreateMessageConsumer.java
使用activemq
public class CreateMessageConsumer {

    @Autowired
    public AgentService agentService;
    
    public void process(String str) {
        System.out.println("system B receive message from  system A ");
        
        agentService.agentPath(str);
    }
}
使用activemq

4.启动系统A和系统B的应用,只要系统A往队列A.CreateMessage产生消息,系统B会自动接收到消息。

相关推荐