使用 ActiveMQ 示例

在 Java 里有 JMS 的多个实现。其中 apache 下的 ActiveMQ 就是不错的选择。还有一个比较热的是 RabbitMQ （是 erlang 语言实现的）。这里示例下使用 ActiveMQ

用 ActiveMQ 最好还是了解下 JMS

JMS 定义了两种方式：Quere（点对点）；Topic（发布/订阅）。

ConnectionFactory 是连接工厂，负责创建Connection。

Connection 负责创建 Session。

Session 创建 MessageProducer（用来发消息） 和 MessageConsumer（用来接收消息）。

Destination 是消息的目的地。

详细的可以网上找些 JMS 规范（有中文版）。

下载 apache-activemq-5.3.0。http://activemq.apache.org/download.html，解压，然后双击 bin/activemq.bat。运行后，可以在 http://localhost:8161/admin 观察。也有 demo， http://localhost:8161/demo。把 activemq-all-5.3.0.jar 加入 classpath。

Jms 发送 代码：

public static void main(String[] args) throws Exception {  
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();  
  
    Connection connection = connectionFactory.createConnection();  
    connection.start();  
  
    Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  
    Destination destination = session.createQueue("my-queue");  
  
    MessageProducer producer = session.createProducer(destination);  
    for(int i=0; i<3; i++) {  
        MapMessage message = session.createMapMessage();  
        message.setLong("count", new Date().getTime());  
        Thread.sleep(1000);  
        //通过消息生产者发出消息  
        producer.send(message);  
    }  
    session.commit();  
    session.close();  
    connection.close();  
}  

Jms 接收代码：

public static void main(String[] args) throws Exception {  
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();  
  
    Connection connection = connectionFactory.createConnection();  
    connection.start();  
  
    final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  
    Destination destination = session.createQueue("my-queue");  
  
    MessageConsumer consumer = session.createConsumer(destination);  
    /*//listener 方式 
    consumer.setMessageListener(new MessageListener() { 
 
        public void onMessage(Message msg) { 
            MapMessage message = (MapMessage) msg; 
            //TODO something.... 
            System.out.println("收到消息：" + new Date(message.getLong("count"))); 
            session.commit(); 
        } 
 
    }); 
    Thread.sleep(30000); 
    */  
    int i=0;  
    while(i<3) {  
        i++;  
        MapMessage message = (MapMessage) consumer.receive();  
        session.commit();  
  
        //TODO something....  
        System.out.println("收到消息：" + new Date(message.getLong("count")));  
    }  
  
    session.close();  
    connection.close();  
}  

启动 JmsReceiver 和 JmsSender 可以在看输出三条时间信息。当然 Jms 还指定有其它格式的数据，如 TextMessage

结合 Spring 的 JmsTemplate 方便用：

xml：

public static void main(String[] args) {  
    ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:app*.xml");  
  
    JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");  
  
    jmsTemplate.send(new MessageCreator() {  
  
        public Message createMessage(Session session) throws JMSException {  
            MapMessage mm = session.createMapMessage();  
            mm.setLong("count", new Date().getTime());  
            return mm;  
        }  
  
    });  
}  

receiver：

public static void main(String[] args) {  
    ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:app*.xml");  
  
    JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");  
    while(true) {  
        Map<String, Object> mm =  (Map<String, Object>) jmsTemplate.receiveAndConvert();  
        System.out.println("收到消息：" + new Date((Long)mm.get("count")));  
    }  
}  

注意：直接用 Jms 接口时接收了消息后要提交一下，否则下次启动接收者时还可以收到旧数据。有了 JmsTemplate 就不用自己提交 session.commit() 了。如果使用了 PooledConnectionFactory 要把 apache-activemq-5.3.0\lib\optional\activemq-pool-5.3.0.jar 加到 classpath