说说在项目里是怎么使用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，

如果页面是下面这样的

     

输入用户名：admin，密码：admin就OK了。

          

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

项目目录结构如下：

　　　　

 

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

DefaultMessageConverter.java

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
     
       map = new HashMap
      
       ();        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
       
         map = (HashMap
        
         ) ((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");        }    }}
        
       
      
     

QueueMessageProducer.java

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);    }}

CreateMessageProducer.java（消息生产者）

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～～～～～～～～～～");    }}

 

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

applicationContext-amq.xml文件

DefaultMessageConverter.java、QueueMessageProducer.java、QueueConsumer.java与系统A一样。 CreateMessageConsumer.java

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);    }}

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