ActiveMQ发布-订阅消息模式（同点对点模式的区别）

2024-01-07 21:21:22

点对点与发布订阅最初是由JMS定义的。这两种模式主要区别或解决的问题就是发送到队列的消息能否重复消费(多订阅)

点对点： 
消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息。这里要注意： 
消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。 
Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。 
发布/订阅 
消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic的消息会被所有订阅者消费。

1、消息生产者-消息发布-Topic

  1. /**
  2. * 消息生产者-消息发布者
  3. * @author Administrator
  4. *
  5. */
  6. public class JMSProducer {
  7. private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
  8. private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
  9. private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址
  10. private static final int SENDNUM=10; // 发送的消息数量
  11. public static void main(String[] args) {
  12. ConnectionFactory connectionFactory; // 连接工厂
  13. Connection connection = null; // 连接
  14. Session session; // 会话 接受或者发送消息的线程
  15. Destination destination; // 消息的目的地
  16. MessageProducer messageProducer; // 消息生产者
  17. // 实例化连接工厂
  18. connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
  19. try {
  20. connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
  21. connection.start(); // 启动连接
  22. session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建Session
  23. // destination=session.createQueue("FirstQueue1"); // 创建消息队列
  24. destination=session.createTopic("FirstTopic1");
  25. messageProducer=session.createProducer(destination); // 创建消息生产者
  26. sendMessage(session, messageProducer); // 发送消息
  27. session.commit();
  28. } catch (Exception e) {
  29. // TODO Auto-generated catch block
  30. e.printStackTrace();
  31. } finally{
  32. if(connection!=null){
  33. try {
  34. connection.close();
  35. } catch (JMSException e) {
  36. // TODO Auto-generated catch block
  37. e.printStackTrace();
  38. }
  39. }
  40. }
  41. }
  42. /**
  43. * 发送消息
  44. * @param session
  45. * @param messageProducer
  46. * @throws Exception
  47. */
  48. public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{
  49. for(int i=0;i<JMSProducer.SENDNUM;i++){
  50. TextMessage message=session.createTextMessage("ActiveMQ 发送的消息"+i);
  51. System.out.println("发送消息："+"ActiveMQ 发布的消息"+i);
  52. messageProducer.send(message);
  53. }
  54. }
  55. }

2、多个消息订阅者-消息消费者

消息订阅者一

  1. /**
  2. * 消息消费者-消息订阅者一
  3. * @author Administrator
  4. *
  5. */
  6. public class JMSConsumer {
  7. private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
  8. private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
  9. private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址
  10. public static void main(String[] args) {
  11. ConnectionFactory connectionFactory; // 连接工厂
  12. Connection connection = null; // 连接
  13. Session session; // 会话 接受或者发送消息的线程
  14. Destination destination; // 消息的目的地
  15. MessageConsumer messageConsumer; // 消息的消费者
  16. // 实例化连接工厂
  17. connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
  18. try {
  19. connection=connectionFactory.createConnection();  // 通过连接工厂获取连接
  20. connection.start(); // 启动连接
  21. session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
  22. // destination=session.createQueue("FirstQueue1");  // 创建连接的消息队列
  23. destination=session.createTopic("FirstTopic1");
  24. messageConsumer=session.createConsumer(destination); // 创建消息消费者
  25. messageConsumer.setMessageListener(new Listener()); // 注册消息监听
  26. } catch (JMSException e) {
  27. // TODO Auto-generated catch block
  28. e.printStackTrace();
  29. }
  30. }
  31. }

消息订阅者二

  1. /**
  2. * 消息消费者-消息订阅者二
  3. * @author Administrator
  4. *
  5. */
  6. public class JMSConsumer2 {
  7. private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
  8. private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
  9. private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址
  10. public static void main(String[] args) {
  11. ConnectionFactory connectionFactory; // 连接工厂
  12. Connection connection = null; // 连接
  13. Session session; // 会话 接受或者发送消息的线程
  14. Destination destination; // 消息的目的地
  15. MessageConsumer messageConsumer; // 消息的消费者
  16. // 实例化连接工厂
  17. connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);
  18. try {
  19. connection=connectionFactory.createConnection();  // 通过连接工厂获取连接
  20. connection.start(); // 启动连接
  21. session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
  22. // destination=session.createQueue("FirstQueue1");  // 创建连接的消息队列
  23. destination=session.createTopic("FirstTopic1");
  24. messageConsumer=session.createConsumer(destination); // 创建消息消费者
  25. messageConsumer.setMessageListener(new Listener2()); // 注册消息监听
  26. } catch (JMSException e) {
  27. // TODO Auto-generated catch block
  28. e.printStackTrace();
  29. }
  30. }
  31. }

两个Linsner用于打印不同的标识信息，故省略。

注：发布订阅模式适用于1个消息生产者，多个消费者场景，首先启动消息订阅方，在消息发布方开始执行后，接收该消息进行处理。在ActiveMQ管理界面会动态跟进消息产生-消费（入队、出队）情况；以及生产者个数，消费者个数。

http://blog.csdn.net/Daybreak1209/article/details/51672277

http://blog.csdn.net/zbw18297786698/article/details/53000605

上一篇：php 事务处理，ActiveMQ的发送消息，与处理消息


下一篇：ActiveMq笔记1-消息可靠性理论