JMS Activemq实战例子demo

Posted 2020-10-29 duanqiao123

上一篇已经讲了JMS的基本概念，这一篇来上手练一练，如果对JMS基本概念还不熟悉，欢迎参靠JMS基本概。

 

这篇文章所使用的代码已经不是我刚入手时的代码，已经经过我重构过的代码，便于理解，并且加了很多中文注释，希望对大家有所帮助。

 

在基本概念一篇中已经讲到，JMS有两种消息模型，一种是点对点，另一种的发布/订阅模式。本篇文章就基于这两种消息模型来写例子。

 

点对点模型

 

        先看一下生产者代码：

 

[java] view plain copy

 

1. package com.darren.activemq.queue;  
2.   
3. import javax.jms.JMSException;  
4. import javax.jms.Session;  
5. import javax.jms.TextMessage;  
6.   
7. import com.darren.activemq.ActivemqContants;  
8. import com.darren.activemq.ProducerConsumer;  
9.   
10. /** 
11.  * 消息生产类 
12.  *  
13.  * @author Darren 
14.  * 
15.  */  
16. public class QueueProducer extends ProducerConsumer {  
17.     private int startNumber;  
18.     private int endNumber;  
19.   
20.     public QueueProducer(String name) throws JMSException {  
21.         this.name = name;  
22.   
23.         // 通过连接工厂获取连接  
24.         this.connection = this.getConnection();  
25.         // 启动连接  
26.         this.connection.start();  
27.         // 创建Session  
28.         this.session = this.connection.createSession(true, Session.AUTO_ACKNOWLEDGE);  
29.         // 创建消息队列  
30.         this.destination = this.session.createQueue("test-queue");  
31.         // 创建消息生产者  
32.         this.messageProducer = this.session.createProducer(destination);  
33.     }  
34.   
35.     /** 
36.      * 发送消息 
37.      *  
38.      * @throws JMSException 
39.      */  
40.     public void sendMessage() throws JMSException {  
41.         this.startNumber = this.endNumber;  
42.         this.endNumber = this.startNumber + MESSAGE_COUNT;  
43.         for (int i = this.startNumber; i < this.endNumber; i++) {  
44.             TextMessage message = this.session.createTextMessage("I send the message " + i);  
45.             System.out.println(message.getText());  
46.             this.messageProducer.send(message);  
47.         }  
48.     }  
49.   
50.     /** 
51.      * 发送结束标志 
52.      *  
53.      * @param times 
54.      *            发送次数 
55.      * @throws JMSException 
56.      */  
57.     public void sendFinishMessage(int times) throws JMSException {  
58.         for (int i = 0; i < times; i++) {  
59.             TextMessage message = this.session.createTextMessage(ActivemqContants.FINISH_FLAG);  
60.             System.out.println("Send finish flag: " + message.getText());  
61.             this.messageProducer.send(message);  
62.         }  
63.     }  
64.   
65.     /** 
66.      * 提交事务 
67.      *  
68.      * @throws JMSException 
69.      */  
70.     public void commit() throws JMSException {  
71.         this.session.commit();  
72.     }  
73. }  

        我写了两种消息消费者，其实也就是上一篇讲的同步消费和异步消费。

 

        先来看异步消费者：

 

 

[java] view plain copy

 

1. package com.darren.activemq.queue;  
2.   
3. import javax.jms.JMSException;  
4. import javax.jms.Session;  
5. import com.darren.activemq.ProducerConsumer;  
6. import com.darren.activemq.listener.ConsumerListener;  
7.   
8. /** 
9.  * 消息消费者 
10.  *  
11.  * @author Darren 
12.  * 
13.  */  
14. public class QueueListenerConsumer extends ProducerConsumer {  
15.   
16.     public QueueListenerConsumer(String name) throws JMSException {  
17.         this.name = name;  
18.   
19.         // 通过连接工厂获取连接  
20.         this.connection = this.getConnection();  
21.         // 启动连接  
22.         this.connection.start();  
23.         // 创建Session  
24.         this.session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
25.         // 创建连接的消息队列  
26.         this.destination = this.session.createQueue("test-queue");  
27.         // 创建消息消费者  
28.         this.messageConsumer = this.session.createConsumer(destination);  
29.         // 设置消息监听  
30.         this.messageConsumer.setMessageListener(new ConsumerListener("Listener 1:", this));  
31.     }  
32. }  

        异步消费者通过设置监听器来达到异步消费消息的目的。其实这个监听器也很简单，我们来看一下。

 

 

[java] view plain copy

 

1. package com.darren.activemq.listener;  
2.   
3. import javax.jms.Message;  
4. import javax.jms.MessageListener;  
5. import javax.jms.ObjectMessage;  
6. import javax.jms.TextMessage;  
7.   
8. import com.darren.activemq.ActivemqContants;  
9. import com.darren.activemq.ProducerConsumer;  
10.   
11. /** 
12.  * 消息监听 
13.  *  
14.  * @author Darren 
15.  * 
16.  */  
17. public class ConsumerListener implements MessageListener {  
18.     private String name;  
19.     private ProducerConsumer producerConsumer;  
20.   
21.     public ConsumerListener(String name, ProducerConsumer producerConsumer) {  
22.         this.name = name;  
23.         this.producerConsumer = producerConsumer;  
24.     }  
25.   
26.     @Override  
27.     public void onMessage(Message message) {  
28.         try {  
29.             if (message instanceof TextMessage) {  
30.                 TextMessage textMessage = (TextMessage) message;  
31.                 System.out.println(name + " 接收到的消息 " + textMessage.getText());  
32.                 // 如果接收到结束标志，修改消费者的状态  
33.                 if (ActivemqContants.FINISH_FLAG.equals(textMessage.getText())) {  
34.                     // 消费者消费完成，关闭连接  
35.                     this.producerConsumer.closeConnection();  
36.                 }  
37.             } else if (message instanceof ObjectMessage) {  
38.                 ObjectMessage objectMessage = (ObjectMessage) message;  
39.   
40.                 System.out.println(name + " 接收到的消息 " + objectMessage.getObject());  
41.             } else {  
42.                 System.out.println("不支持的消息类型！");  
43.             }  
44.         } catch (Exception e) {  
45.             e.printStackTrace();  
46.         }  
47.   
48.     }  
49.   
50. }  

        这里只有一个方法。

 

        为了便于理解，我先把父类贴出来，这样就可以基本完整的看了。

 

 

[java] view plain copy

 

1. package com.darren.activemq;  
2.   
3. import javax.jms.Connection;  
4. import javax.jms.ConnectionFactory;  
5. import javax.jms.Destination;  
6. import javax.jms.JMSException;  
7. import javax.jms.MessageConsumer;  
8. import javax.jms.MessageProducer;  
9. import javax.jms.Session;  
10.   
11. import org.apache.activemq.ActiveMQConnection;  
12. import org.apache.activemq.ActiveMQConnectionFactory;  
13.   
14. public abstract class ProducerConsumer {  
15.     protected static final String USERNAME = ActiveMQConnection.DEFAULT_USER;// 默认的连接用户名  
16.     protected static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;// 默认的连接密码  
17.     protected static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL;// 默认的连接地址  
18.     protected static final int MESSAGE_COUNT = 10;// 发送的消息数量  
19.   
20.     protected static ConnectionFactory connectionFactory;// 连接工厂  
21.     protected Connection connection = null; // 连接  
22.     protected Session session;// 会话 接受或者发送消息的线程  
23.     protected Destination destination;// 消息的目的地  
24.     protected MessageProducer messageProducer;// 消息生产者  
25.     protected MessageConsumer messageConsumer;// 消息消费者  
26.   
27.     // 状态  
28.     protected volatile boolean isFinished = false;  
29.     protected String name;  
30.   
31.     static {  
32.         getConnectionFactory();  
33.     }  
34.   
35.     /** 
36.      * 获取连接工厂 
37.      *  
38.      * @return 
39.      */  
40.     protected synchronized static ConnectionFactory getConnectionFactory() {  
41.         if (connectionFactory == null) {  
42.             connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKERURL);  
43.         }  
44.   
45.         return connectionFactory;  
46.     }  
47.   
48.     /** 
49.      * 获取连接 
50.      *  
51.      * @return 
52.      * @throws JMSException 
53.      */  
54.     protected Connection getConnection() throws JMSException {  
55.         return connectionFactory.createConnection();  
56.     };  
57.   
58.     /** 
59.      * 关闭连接 
60.      *  
61.      * @throws JMSException 
62.      */  
63.     public void closeConnection() throws JMSException {  
64.         if (this.connection != null) {  
65.             this.connection.close();  
66.         }  
67.   
68.         System.out.println(this.name + " connection close!");  
69.   
70.         this.isFinished = true;  
71.     }  
72.   
73.     /** 
74.      * 获取状态 
75.      *  
76.      * @return 
77.      */  
78.     public boolean isFinished() {  
79.         return isFinished;  
80.     }  
81.   
82.     /** 
83.      * 获取消费者 
84.      *  
85.      * @return 
86.      */  
87.     public MessageConsumer getMessageConsumer() {  
88.         return messageConsumer;  
89.     }  
90.   
91.     /** 
92.      * 获取生产者或消费者名称 
93.      *  
94.      * @return 
95.      */  
96.     public String getName() {  
97.         return name;  
98.     }  
99. }  

        用到的常量类

 

 

[java] view plain copy

 

1. package com.darren.activemq;  
2.   
3. public class ActivemqContants {  
4.     public static final String FINISH_FLAG = "FINISHED";  
5. }  

 

        来个测试类吧，这样就完整了

 

[java] view plain copy

 

1. package com.darren.activemq.queue;  
2.   
3. import javax.jms.JMSException;  
4.   
5. public class QueueTest {  
6.     public static void main(String[] args) {  
7.         Thread thread = null;  
8.         try {  
9.             // 启动消费者，消费者开始等待  
10.             new QueueListenerConsumer("QueueListenerConsumer");  
11.             //new QueueReceiveConsumer("QueueReceiveConsumer");  
12.   
13.             thread = new Thread(new Runnable() {  
14.   
15.                 @Override  
16.                 public void run() {  
17.                     try {  
18.                         // 启动生产者，生产者定时生产消息  
19.                         QueueProducer producer = new QueueProducer("QueueProducer");  
20.                         Thread.sleep(2000);  
21.                         // 第一次发送  
22.                         producer.sendMessage();  
23.                         producer.commit();  
24.   
25.                         Thread.sleep(2000);  
26.                         // 第二次发送  
27.                         producer.sendMessage();  
28.                         producer.commit();  
29.   
30.                         Thread.sleep(2000);  
31.                         // 第三次发送  
32.                         producer.sendMessage();  
33.                         producer.commit();  
34.   
35.                         // 发送结束标志  
36.                         producer.sendFinishMessage(1);  
37.                         producer.commit();  
38.   
39.                         // 生产者生产完成，关闭连接  
40.                         producer.closeConnection();  
41.                     } catch (Exception e) {  
42.                         e.printStackTrace();  
43.                     }  
44.                 }  
45.             });  
46.             thread.start();  
47.         } catch (JMSException e) {  
48.             e.printStackTrace();  
49.         }  
50.     }  
51. }  

 

        现在为止，一个异步消息的完整例子就真的完整了，我是为了便于看到消息的消费过程，所以把生产者放到了一个线程里来不断的生产消息，来通过观察消费者打印的消费信息，了解消费的过程。

 

        再来看看同步消息模式：

        同步消费的方式就是轮训的去要有没有消息，有没有消息。为了适应我的测试类，我把同步消息消费也按照异步消费的模式做了相应的改动，下边就是代码：

 

[java] view plain copy

 

1. package com.darren.activemq.queue;  
2.   
3. import javax.jms.JMSException;  
4. import javax.jms.Session;  
5. import com.darren.activemq.ProducerConsumer;  
6. import com.darren.activemq.listener.UglyConsumerListener;  
7.   
8. /** 
9.  * 消息消费者 
10.  *  
11.  * @author Darren 
12.  * 
13.  */  
14. public class QueueReceiveConsumer extends ProducerConsumer {  
15.   
16.     public QueueReceiveConsumer(String name) throws JMSException {  
17.         this.name = name;  
18.   
19.         // 通过连接工厂获取连接  
20.         this.connection = this.getConnection();  
21.         // 启动连接  
22.         this.connection.start();  
23.         // 创建Session  
24.         this.session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
25.         // 创建连接的消息队列  
26.         this.destination = this.session.createQueue("test-queue");  
27.         // 创建消息消费者  
28.         this.messageConsumer = this.session.createConsumer(destination);  
29.   
30.         // 启动一个异步线程接受消息，模拟一个消息监听器  
31.         Thread thread = new Thread(new UglyConsumerListener("UglyListener 1:", this));  
32.         thread.start();  
33.     }  
34. }  

 

        在这里，我选择启动一个线程去设置一个监听器，为什么要启动线程，因为我上边提到了要适应我的测试类，我是先初始化消费者，如果不用线程，那么消费者就会在要消息的时候夯住，因为：

 

[java] view plain copy

 

1. Message message = messageConsumer.receive(100000);  

 

 

        这个方法是阻塞式方法。会程序在初始化消费者后就会夯住，然后生成者无法启动，无法生成消息，整个程序就死锁了，为了解决这个问题，所以采用模拟异步消费的方式来写同步消费例子。

        下边来看我写的丑陋的监听器：

 

[java] view plain copy

 

1. package com.darren.activemq.listener;  
2.   
3. import javax.jms.JMSException;  
4. import javax.jms.Message;  
5. import javax.jms.MessageConsumer;  
6. import javax.jms.ObjectMessage;  
7. import javax.jms.TextMessage;  
8. import com.darren.activemq.ActivemqContants;  
9. import com.darren.activemq.ProducerConsumer;  
10.   
11. /** 
12.  * 消息监听 
13.  *  
14.  * @author Darren 
15.  * 
16.  */  
17. public class UglyConsumerListener implements Runnable {  
18.     private String name;  
19.     private ProducerConsumer producerConsumer;  
20.   
21.     public UglyConsumerListener(String name, ProducerConsumer producerConsumer) {  
22.         this.name = name;  
23.         this.producerConsumer = producerConsumer;  
24.     }  
25.   
26.     public void onMessage(Message message) {  
27.         try {  
28.             if (message instanceof TextMessage) {  
29.                 TextMessage textMessage = (TextMessage) message;  
30.                 System.out.println(name + " 接收到的消息 " + textMessage.getText());  
31.                 // 如果接收到结束标志，修改消费者的状态  
32.                 if (ActivemqContants.FINISH_FLAG.equals(textMessage.getText())) {  
33.                     // 消费者消费完成，关闭连接  
34.                     this.producerConsumer.closeConnection();  
35.                 }  
36.             } else if (message instanceof ObjectMessage) {  
37.                 ObjectMessage objectMessage = (ObjectMessage) message;  
38.   
39.                 System.out.println(name + " 接收到的消息 " + objectMessage.getObject());  
40.             } else {  
41.                 System.out.println("不支持的消息类型！");  
42.             }  
43.         } catch (Exception e) {  
44.             e.printStackTrace();  
45.         }  
46.   
47.     }  
48.   
49.     @Override  
50.     public void run() {  
51.         while (!producerConsumer.isFinished()) {  
52.             MessageConsumer messageConsumer = this.producerConsumer.getMessageConsumer();  
53.             if (messageConsumer != null) {  
54.                 try {  
55.                     Message message = messageConsumer.receive(100000);  
56.                     this.onMessage(message);  
57.                 } catch (JMSException e) {  
58.                     e.printStackTrace();  
59.                 }  
60.             }  
61.         }  
62.     }  
63. }  

        为什么称为丑陋的监听器，相信大家也不难发现，因为采用循环方式，并不是真正的监听器。 看了这两个监听器，那么我提出一个问题，这个问题也是在我写这篇文章的时候发现的，如果我再写一个抽象类去作为这两个监听器的父类，把onMessage方法提出来，是不是更加优化了，因为我发现两个onMessage方法其实一样，我也是故意给做的一样的。

 

        当然今天讲的并不是Java的重构，而是JMS，这个问题就作为题外话好了。

 

        现在你自己也可以去跑一跑这两个例子，细心的同学或发现在测试类中其实有一行有疑问。为什么发送结束消息的方法要加参数，可以设置发送多个结束消息。这个问题我们稍后介绍，下面我来运行一下代码，也许你就渐渐明白。

 

        先跑一个异步消费的例子：

 

[plain] view plain copy

 

1.  INFO | Successfully connected to tcp://localhost:61616  
2.  INFO | Successfully connected to tcp://localhost:61616  
3. I send the message 0  
4. I send the message 1  
5. I send the message 2  
6. I send the message 3  
7. I send the message 4  
8. I send the message 5  
9. I send the message 6  
10. I send the message 7  
11. I send the message 8  
12. I send the message 9  
13. Listener 1: 接收到的消息 I send the message 0  
14. Listener 1: 接收到的消息 I send the message 1  
15. Listener 1: 接收到的消息 I send the message 2  
16. Listener 1: 接收到的消息 I send the message 3  
17. Listener 1: 接收到的消息 I send the message 4  
18. Listener 1: 接收到的消息 I send the message 5  
19. Listener 1: 接收到的消息 I send the message 6  
20. Listener 1: 接收到的消息 I send the message 7  
21. Listener 1: 接收到的消息 I send the message 8  
22. Listener 1: 接收到的消息 I send the message 9  
23. I send the message 10  
24. I send the message 11  
25. I send the message 12  
26. I send the message 13  
27. I send the message 14  
28. I send the message 15  
29. I send the message 16  
30. I send the message 17  
31. I send the message 18  
32. I send the message 19  
33. Listener 1: 接收到的消息 I send the message 10  
34. Listener 1: 接收到的消息 I send the message 11  
35. Listener 1: 接收到的消息 I send the message 12  
36. Listener 1: 接收到的消息 I send the message 13  
37. Listener 1: 接收到的消息 I send the message 14  
38. Listener 1: 接收到的消息 I send the message 15  
39. Listener 1: 接收到的消息 I send the message 16  
40. Listener 1: 接收到的消息 I send the message 17  
41. Listener 1: 接收到的消息 I send the message 18  
42. Listener 1: 接收到的消息 I send the message 19  
43. I send the message 20  
44. I send the message 21  
45. I send the message 22  
46. I send the message 23  
47. I send the message 24  
48. I send the message 25  
49. I send the message 26  
50. I send the message 27  
51. I send the message 28  
52. I send the message 29  
53. Listener 1: 接收到的消息 I send the message 20  
54. Listener 1: 接收到的消息 I send the message 21  
55. Send finish flag: FINISHED  
56. Listener 1: 接收到的消息 I send the message 22  
57. Listener 1: 接收到的消息 I send the message 23  
58. Listener 1: 接收到的消息 I send the message 24  
59. Listener 1: 接收到的消息 I send the message 25  
60. Listener 1: 接收到的消息 I send the message 26  
61. Listener 1: 接收到的消息 I send the message 27  
62. Listener 1: 接收到的消息 I send the message 28  
63. Listener 1: 接收到的消息 I send the message 29  
64. Listener 1: 接收到的消息 FINISHED  
65. QueueProducer connection close!  
66. QueueListenerConsumer connection close!  

        然后再来跑一个同步消费的例子：

 

 

[java] view plain copy

 

1. //new QueueListenerConsumer("QueueListenerConsumer");  
2. new QueueReceiveConsumer("QueueReceiveConsumer");  

        只需把注水换一换就行了。

 

 

[plain] view plain copy

 

1.  INFO | Successfully connected to tcp://localhost:61616  
2.  INFO | Successfully connected to tcp://localhost:61616  
3. I send the message 0  
4. I send the message 1  
5. I send the message 2  
6. I send the message 3  
7. I send the message 4  
8. I send the message 5  
9. I send the message 6  
10. I send the message 7  
11. I send the message 8  
12. I send the message 9  
13. UglyListener 1: 接收到的消息 I send the message 0  
14. UglyListener 1: 接收到的消息 I send the message 1  
15. UglyListener 1: 接收到的消息 I send the message 2  
16. UglyListener 1: 接收到的消息 I send the message 3  
17. UglyListener 1: 接收到的消息 I send the message 4  
18. UglyListener 1: 接收到的消息 I send the message 5  
19. UglyListener 1: 接收到的消息 I send the message 6  
20. UglyListener 1: 接收到的消息 I send the message 7  
21. UglyListener 1: 接收到的消息 I send the message 8  
22. UglyListener 1: 接收到的消息 I send the message 9  
23. I send the message 10  
24. I send the message 11  
25. I send the message 12  
26. I send the message 13  
27. I send the message 14  
28. I send the message 15  
29. I send the message 16  
30. I send the message 17  
31. I send the message 18  
32. I send the message 19  
33. UglyListener 1: 接收到的消息 I send the message 10  
34. UglyListener 1: 接收到的消息 I send the message 11  
35. UglyListener 1: 接收到的消息 I send the message 12  
36. UglyListener 1: 接收到的消息 I send the message 13  
37. UglyListener 1: 接收到的消息 I send the message 14  
38. UglyListener 1: 接收到的消息 I send the message 15  
39. UglyListener 1: 接收到的消息 I send the message 16  
40. UglyListener 1: 接收到的消息 I send the message 17  
41. UglyListener 1: 接收到的消息 I send the message 18  
42. UglyListener 1: 接收到的消息 I send the message 19  
43. I send the message 20  
44. I send the message 21  
45. I send the message 22  
46. I send the message 23  
47. I send the message 24  
48. I send the message 25  
49. I send the message 26  
50. I send the message 27  
51. I send the message 28  
52. I send the message 29  
53. Send finish flag: FINISHED  
54. UglyListener 1: 接收到的消息 I send the message 20  
55. UglyListener 1: 接收到的消息 I send the message 21  
56. UglyListener 1: 接收到的消息 I send the message 22  
57. UglyListener 1: 接收到的消息 I send the message 23  
58. UglyListener 1: 接收到的消息 I send the message 24  
59. UglyListener 1: 接收到的消息 I send the message 25  
60. UglyListener 1: 接收到的消息 I send the message 26  
61. UglyListener 1: 接收到的消息 I send the message 27  
62. UglyListener 1: 接收到的消息 I send the message 28  
63. UglyListener 1: 接收到的消息 I send the message 29  
64. UglyListener 1: 接收到的消息 FINISHED  
65. QueueReceiveConsumer connection close!  
66. QueueProducer connection close!  

        大家会发现和异步消费并未区别，因为其实也是异步消费。

 

        那么如果我现在把测试类的注释都打开，会是什么结果呢：

 

[java] view plain copy

 

1. new QueueListenerConsumer("QueueListenerConsumer");  
2. new QueueReceiveConsumer("QueueReceiveConsumer");  

 

[plain] view plain copy

 

1.  INFO | Successfully connected to tcp://localhost:61616  
2.  INFO | Successfully connected to tcp://localhost:61616  
3.  INFO | Successfully connected to tcp://localhost:61616  
4. I send the message 0  
5. I send the message 1  
6. I send the message 2  
7. I send the message 3  
8. I send the message 4  
9. I send the message 5  
10. I send the message 6  
11. I send the message 7  
12. I send the message 8  
13. I send the message 9  
14. Listener 1: 接收到的消息 I send the message 0  
15. Listener 1: 接收到的消息 I send the message 2  
16. UglyListener 1: 接收到的消息 I send the message 1  
17. Listener 1: 接收到的消息 I send the message 4  
18. Listener 1: 接收到的消息 I send the message 6  
19. Listener 1: 接收到的消息 I send the message 8  
20. UglyListener 1: 接收到的消息 I send the message 3  
21. UglyListener 1: 接收到的消息 I send the message 5  
22. UglyListener 1: 接收到的消息 I send the message 7  
23. UglyListener 1: 接收到的消息 I send the message 9  
24. I send the message 10  
25. I send the message 11  
26. I send the message 12  
27. I send the message 13  
28. I send the message 14  
29. I send the message 15  
30. I send the message 16  
31. I send the message 17  
32. I send the message 18  
33. I send the message 19  
34. Listener 1: 接收到的消息 I send the message 10  
35. Listener 1: 接收到的消息 I send the message 12  
36. Listener 1: 接收到的消息 I send the message 14  
37. UglyListener 1: 接收到的消息 I send the message 11  
38. Listener 1: 接收到的消息 I send the message 16  
39. UglyListener 1: 接收到的消息 I send the message 13  
40. Listener 1: 接收到的消息 I send the message 18  
41. UglyListener 1: 接收到的消息 I send the message 15  
42. UglyListener 1: 接收到的消息 I send the message 17  
43. UglyListener 1: 接收到的消息 I send the message 19  
44. I send the message 20  
45. I send the message 21  
46. I send the message 22  
47. I send the message 23  
48. I send the message 24  
49. I send the message 25  
50. I send the message 26  
51. I send the message 27  
52. I send the message 28  
53. I send the message 29  
54. Listener 1: 接收到的消息 I send the message 20  
55. Listener 1: 接收到的消息 I send the message 22  
56. Listener 1: 接收到的消息 I send the message 24  
57. Listener 1: 接收到的消息 I send the message 26  
58. Listener 1: 接收到的消息 I send the message 28  
59. Send finish flag: FINISHED  
60. UglyListener 1: 接收到的消息 I send the message 21  
61. UglyListener 1: 接收到的消息 I send the message 23  
62. UglyListener 1: 接收到的消息 I send the message 25  
63. UglyListener 1: 接收到的消息 I send the message 27  
64. UglyListener 1: 接收到的消息 I send the message 29  
65. Listener 1: 接收到的消息 FINISHED  
66. QueueProducer connection close!  
67. QueueListenerConsumer connection close!  

        这个时候有俩个消费者，一个消费者消费了一半的消息，这里我没有深入研究这种消费平均的方式是随机的还是真的平均，我猜猜可能是真的平均，因为我运行了多次都是同样的结果。

 

        通过我打出来的消息，发现只有一个消费者结束了，另一个并没有结束，因为没有收到结束消息标志，所以需要发送两个结束标志才能使两个消费者都能结束。

 

[java] view plain copy

 

1. producer.sendFinishMessage(2);  

        这也正是我设置参数的原因：

 

        改成2之后再来运行一次：

 

[plain] view plain copy

 

1.  INFO | Successfully connected to tcp://localhost:61616  
2.  INFO | Successfully connected to tcp://localhost:61616  
3.  INFO | Successfully connected to tcp://localhost:61616  
4. I send the message 0  
5. I send the message 1  
6. I send the message 2  
7. I send the message 3  
8. I send the message 4  
9. I send the message 5  
10. I send the message 6  
11. I send the message 7  
12. I send the message 8  
13. I send the message 9  
14. Listener 1: 接收到的消息 I send the message 0  
15. Listener 1: 接收到的消息 I send the message 2  
16. UglyListener 1: 接收到的消息 I send the message 1  
17. Listener 1: 接收到的消息 I send the message 4  
18. UglyListener 1: 接收到的消息 I send the message 3  
19. Listener 1: 接收到的消息 I send the message 6  
20. UglyListener 1: 接收到的消息 I send the message 5  
21. Listener 1: 接收到的消息 I send the message 8  
22. UglyListener 1: 接收到的消息 I send the message 7  
23. UglyListener 1: 接收到的消息 I send the message 9  
24. I send the message 10  
25. I send the message 11  
26. I send the message 12  
27. I send the message 13  
28. I send the message 14  
29. I send the message 15  
30. I send the message 16  
31. I send the message 17  
32. I send the message 18  
33. I send the message 19  
34. Listener 1: 接收到的消息 I send the message 10  
35. Listener 1: 接收到的消息 I send the message 12  
36. Listener 1: 接收到的消息 I send the message 14  
37. Listener 1: 接收到的消息 I send the message 16  
38. Listener 1: 接收到的消息 I send the message 18  
39. UglyListener 1: 接收到的消息 I send the message 11  
40. UglyListener 1: 接收到的消息 I send the message 13  
41. UglyListener 1: 接收到的消息 I send the message 15  
42. UglyListener 1: 接收到的消息 I send the message 17  
43. UglyListener 1: 接收到的消息 I send the message 19  
44. I send the message 20  
45. I send the message 21  
46. I send the message 22  
47. I send the message 23  
48. I send the message 24  
49. I send the message 25  
50. I send the message 26  
51. I send the message 27  
52. I send the message 28  
53. I send the message 29  
54. Listener 1: 接收到的消息 I send the message 20  
55. Listener 1: 接收到的消息 I send the message 22  
56. Listener 1: 接收到的消息 I send the message 24  
57. Listener 1: 接收到的消息 I send the message 26  
58. Listener 1: 接收到的消息 I send the message 28  
59. Send finish flag: FINISHED  
60. Send finish flag: FINISHED  
61. UglyListener 1: 接收到的消息 I send the message 21  
62. UglyListener 1: 接收到的消息 I send the message 23  
63. UglyListener 1: 接收到的消息 I send the message 25  
64. UglyListener 1: 接收到的消息 I send the message 27  
65. UglyListener 1: 接收到的消息 I send the message 29  
66. Listener 1: 接收到的消息 FINISHED  
67. UglyListener 1: 接收到的消息 FINISHED  
68. QueueProducer connection close!  
69. QueueReceiveConsumer connection close!  
70. QueueListenerConsumer connection close!  

        现在两个消费者都正常结束了。到此我的点对点模式就介绍完了。

 

 

订阅/发布模式

 

        生产者：

 

[java] view plain copy

 

1. package com.darren.activemq.topic;  
2.   
3. import javax.jms.JMSException;  
4. import javax.jms.Session;  
5. import javax.jms.TextMessage;  
6.   
7. import com.darren.activemq.ActivemqContants;  
8. import com.darren.activemq.ProducerConsumer;  
9.   
10. public class TopicProducer extends ProducerConsumer {  
11.     private int startNumber;  
12.     private int endNumber;  
13.   
14.     public TopicProducer(String name) throws JMSException {  
15.         this.name = name;  
16.   
17.         // 通过连接工厂获取连接  
18.         this.connection = this.getConnection();  
19.         // 启动连接  
20.         this.connection.start();  
21.         // 创建Session  
22.         this.session = this.connection.createSession(true, Session.AUTO_ACKNOWLEDGE);  
23.         // 创建消息队列  
24.         this.destination = this.session.createTopic("test-topic");  
25.         // 创建消息生产者  
26.         this.messageProducer = this.session.createProducer(destination);  
27.     }  
28.   
29.     /** 
30.      * 发送消息 
31.      *  
32.      * @throws JMSException 
33.      */  
34.     public void sendMessage() throws JMSException {  
35.         this.startNumber = this.endNumber;  
36.         this.endNumber = this.startNumber + MESSAGE_COUNT;  
37.         for (int i = this.startNumber; i < this.endNumber; i++) {  
38.             TextMessage message = this.session.createTextMessage("I send the message " + i);  
39.             System.out.println(message.getText());  
40.             this.messageProducer.send(message);  
41.         }  
42.     }  
43.   
44.     /** 
45.      * 发送结束标志 
46.      *  
47.      * @throws JMSException 
48.      */  
49.     public void sendFinishMessage() throws JMSException {  
50.         TextMessage message = this.session.createTextMessage(ActivemqContants.FINISH_FLAG);  
51.         System.out.println("Send finish flag: " + message.getText());  
52.         this.messageProducer.send(message);  
53.     }  
54.   
55.     /** 
56.      * 提交事务 
57.      *  
58.      * @throws JMSException 
59.      */  
60.     public void commit() throws JMSException {  
61.         this.session.commit();  
62.     }  
63. }  

        消费者：

 

 

[java] view plain copy

 

1. package com.darren.activemq.topic;  
2.   
3. import javax.jms.JMSException;  
4. import javax.jms.Session;  
5. import com.darren.activemq.ProducerConsumer;  
6. import com.darren.activemq.listener.ConsumerListener;  
7.   
8. public class TopicListenerConsumer extends ProducerConsumer {  
9.   
10.     public TopicListenerConsumer(String name) throws JMSException {  
11.         this.name = name;  
12.   
13.         // 通过连接工厂获取连接  
14.         this.connection = this.getConnection();  
15.         // 启动连接  
16.         this.connection.start();  
17.         // 创建Session  
18.         this.session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
19.         // 创建连接的消息队列  
20.         this.destination = this.session.createTopic("test-topic");  
21.         // 创建消息消费者  
22.         this.messageConsumer = this.session.createConsumer(destination);  
23.         // 设置消息监听  
24.         this.messageConsumer.setMessageListener(new ConsumerListener("Listener 1:", this));  
25.     }  
26. }  

 

[plain] view plain copy

 

1. package com.darren.activemq.topic;  
2.   
3. import javax.jms.JMSException;  
4. import javax.jms.Session;  
5. import com.darren.activemq.ProducerConsumer;  
6. import com.darren.activemq.listener.UglyConsumerListener;  
7.   
8. public class TopicReceiveConsumer extends ProducerConsumer {  
9.   
10.     public TopicReceiveConsumer(String name) throws JMSException {  
11.         this.name = name;  
12.   
13.         // 通过连接工厂获取连接  
14.         this.connection = this.getConnection();  
15.         // 启动连接  
16.         this.connection.start();  
17.         // 创建Session  
18.         this.session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
19.         // 创建连接的消息队列  
20.         this.destination = this.session.createTopic("test-topic");  
21.         // 创建消息消费者  
22.         this.messageConsumer = this.session.createConsumer(destination);  
23.   
24.         // 启动一个异步线程接受消息，模拟一个消息监听器  
25.         Thread thread = new Thread(new UglyConsumerListener("UglyListener 1:", this));  
26.         thread.start();  
27.     }  
28. }  

        大家可以看出相比点对点模式的例子，消息的订阅/发布模式只做了很小的改动：

 

 

[java] view plain copy

 

1. this.destination = this.session.createQueue("test-queue");  
2. //改成  
3. this.destination = this.session.createTopic("test-topic");  

        其他代码都重用了，接下来是测试类：

 

 

[java] view plain copy

 

1. package com.darren.activemq.topic;  
2.   
3. import javax.jms.JMSException;  
4.   
5. public class TopicTest {  
6.     public static void main(String[] args) {  
7.         Thread thread = null;  
8.         try {  
9.             // 启动消费者，消费者开始等待  
10.             new TopicListenerConsumer("TopicListenerConsumer");  
11.             new TopicReceiveConsumer("TopicReceiveConsumer");  
12.   
13.             thread = new Thread(new Runnable() {  
14.   
15.                 @Override  
16.                 public void run() {  
17.                     try {  
18.                         // 启动生产者，生产者定时生产消息  
19.                         TopicProducer producer = new TopicProducer("TopicProducer");  
20.                         Thread.sleep(2000);  
21.                         // 第一次发送  
22.                         producer.sendMessage();  
23.                         producer.commit();  
24.   
25.                         Thread.sleep(2000);  
26.                         // 第二次发送  
27.                         producer.sendMessage();  
28.                         producer.commit();  
29.   
30.                         Thread.sleep(2000);  
31.                         // 第三次发送  
32.                         producer.sendMessage();  
33.                         producer.commit();  
34.   
35.                         // 发送结束标志  
36.                         producer.sendFinishMessage();  
37.                         producer.commit();  
38.   
39.                         // 生产者生产完成，关闭连接  
40.                         producer.closeConnection();  
41.                     } catch (Exception e) {  
42.                         e.printStackTrace();  
43.                     }  
44.                 }  
45.             });  
46.             thread.start();  
47.         } catch (JMSException e) {  
48.             e.printStackTrace();  
49.         }  
50.     }  
51. }  

        运行这个例子：

 

 

[plain] view plain copy

 

1.  INFO | Successfully connected to tcp://localhost:61616  
2.  INFO | Successfully connected to tcp://localhost:61616  
3.  INFO | Successfully connected to tcp://localhost:61616  
4. I send the message 0  
5. I send the message 1  
6. I send the message 2  
7. I send the message 3  
8. I send the message 4  
9. I send the message 5  
10. I send the message 6  
11. I send the message 7  
12. I send the message 8  
13. I send the message 9  
14. Listener 1: 接收到的消息 I send the message 0  
15. Listener 1: 接收到的消息 I send the message 1  
16. UglyListener 1: 接收到的消息 I send the message 0  
17. Listener 1: 接收到的消息 I send the message 2  
18. UglyListener 1: 接收到的消息 I send the message 1  
19. Listener 1: 接收到的消息 I send the message 3  
20. Listener 1: 接收到的消息 I send the message 4  
21. UglyListener 1: 接收到的消息 I send the message 2  
22. Listener 1: 接收到的消息 I send the message 5  
23. UglyListener 1: 接收到的消息 I send the message 3  
24. Listener 1: 接收到的消息 I send the message 6  
25. Listener 1: 接收到的消息 I send the message 7  
26. UglyListener 1: 接收到的消息 I send the message 4  
27. Listener 1: 接收到的消息 I send the message 8  
28. UglyListener 1: 接收到的消息 I send the message 5  
29. Listener 1: 接收到的消息 I send the message 9  
30. UglyListener 1: 接收到的消息 I send the message 6  
31. UglyListener 1: 接收到的消息 I send the message 7  
32. UglyListener 1: 接收到的消息 I send the message 8  
33. UglyListener 1: 接收到的消息 I send the message 9  
34. I send the message 10  
35. I send the message 11  
36. I send the message 12  
37. I send the message 13  
38. I send the message 14  
39. I send the message 15  
40. I send the message 16  
41. I send the message 17  
42. I send the message 18  
43. I send the message 19  
44. Listener 1: 接收到的消息 I send the message 10  
45. Listener 1: 接收到的消息 I send the message 11  
46. Listener 1: 接收到的消息 I send the message 12  
47. Listener 1: 接收到的消息 I send the message 13  
48. Listener 1: 接收到的消息 I send the message 14  
49. Listener 1: 接收到的消息 I send the message 15  
50. Listener 1: 接收到的消息 I send the message 16  
51. Listener 1: 接收到的消息 I send the message 17  
52. Listener 1: 接收到的消息 I send the message 18  
53. Listener 1: 接收到的消息 I send the message 19  
54. UglyListener 1: 接收到的消息 I send the message 10  
55. UglyListener 1: 接收到的消息 I send the message 11  
56. UglyListener 1: 接收到的消息 I send the message 12  
57. UglyListener 1: 接收到的消息 I send the message 13  
58. UglyListener 1: 接收到的消息 I send the message 14  
59. UglyListener 1: 接收到的消息 I send the message 15  
60. UglyListener 1: 接收到的消息 I send the message 16  
61. UglyListener 1: 接收到的消息 I send the message 17  
62. UglyListener 1: 接收到的消息 I send the message 18  
63. UglyListener 1: 接收到的消息 I send the message 19  
64. I send the message 20  
65. I send the message 21  
66. I send the message 22  
67. I send the message 23  
68. I send the message 24  
69. I send the message 25  
70. I send the message 26  
71. I send the message 27  
72. I send the message 28  
73. I send the message 29  
74. Listener 1: 接收到的消息 I send the message 20  
75. Listener 1: 接收到的消息 I send the message 21  
76. Listener 1: 接收到的消息 I send the message 22  
77. Listener 1: 接收到的消息 I send the message 23  
78. Listener 1: 接收到的消息 I send the message 24  
79. Listener 1: 接收到的消息 I send the message 25  
80. Listener 1: 接收到的消息 I send the message 26  
81. Listener 1: 接收到的消息 I send the message 27  
82. Send finish flag: FINISHED  
83. UglyListener 1: 接收到的消息 I send the message 20  
84. UglyListener 1: 接收到的消息 I send the message 21  
85. Listener 1: 接收到的消息 I send the message 28  
86. UglyListener 1: 接收到的消息 I send the message 22  
87. Listener 1: 接收到的消息 I send the message 29  
88. UglyListener 1: 接收到的消息 I send the message 23  
89. UglyListener 1: 接收到的消息 I send the message 24  
90. UglyListener 1: 接收到的消息 I send the message 25  
91. UglyListener 1: 接收到的消息 I send the message 26  
92. Listener 1: 接收到的消息 FINISHED  
93. UglyListener 1: 接收到的消息 I send the message 27  
94. UglyListener 1: 接收到的消息 I send the message 28  
95. UglyListener 1: 接收到的消息 I send the message 29  
96. UglyListener 1: 接收到的消息 FINISHED  
97. TopicListenerConsumer connection close!  
98. TopicReceiveConsumer connection close!  
99. TopicProducer connection close!  

        两个订阅者获取的消息一模一样，并不是平半分了消息，这就是订阅/发布与点对点的不同。

 

        注：订阅/发布模式必须要先订阅，这样订阅者才能收到消息。

 

完