【Active入门-3】ActiveMQ学习-发布者与订阅者

2015年4月28日
1个发布者，1个订阅者，topic
方式1：
先发布消息：

然后订阅消息：

方式2：
先订阅消息：

然后发布消息：

订阅者如下：

结论1:
从上面可以看出，消息发布需要在线发布。


1个发布者，2个订阅者，topic
方式1：
先发布消息：
然后开启两个订阅者
方式2：
先开启两个订阅者：

然后发布消息：

看到订阅者如下：

方式3：
先发布消息，发布完消息后，让发布者休眠10s，然后退出：
此时开启订阅者：

直到发布者程序退出：

订阅者也没有订阅到任何消息
方式4：

上图中，消息1发布完成后，打开订阅者1和订阅者2，紧接着发布者又发布消息2；
在发布者发送消息3之前，订阅者1和订阅者2运行结束。

结论:
1.  在线发布消息：
        订阅者能订阅到消息的条件是：
        1.1  订阅者程序先运行，即在发布者程序发布消息时，订阅者程序处于运行状态（在线状态）；
        1.2  由方式3可以看到，发布者程序处于运行状态，并不代表发布者程序处于发布消息状态；
2. 发布者发布的消息，可以被所有订阅者订阅，且所有订阅者不存在竞争关系。



参考程序：需要略作修改，以实现上面提到的方式1.方式2....
订阅者程序：2015年4月28日 9:57:40
  1. package com.ll.activemq;
  3. import java.util.Collection;
  4. import java.util.Iterator;
  6. import javax.jms.Connection;
  7. import javax.jms.ConnectionFactory;
  8. import javax.jms.Destination;
  9. import javax.jms.MapMessage;
  10. import javax.jms.MessageConsumer;
  11. import javax.jms.Session;
  12. import javax.jms.TextMessage;
  13. import org.apache.activemq.ActiveMQConnection;
  14. import org.apache.activemq.ActiveMQConnectionFactory;
  16. public class Receiver {
  17. 	public static void main(String[] args) {
  18. 		// ConnectionFactory ：连接工厂，JMS 用它创建连接
  19. 		ConnectionFactory connectionFactory;
  20. 		// Connection ：JMS 客户端到JMS Provider 的连接
  21. 		Connection connection = null;
  22. 		// Session： 一个发送或接收消息的线程
  23. 		Session session;
  24. 		// Destination ：消息的目的地;消息发送给谁.
  25. 		Destination destination;
  26. 		// 消费者，消息接收者
  27. 		MessageConsumer consumer;
  28. 		connectionFactory = new ActiveMQConnectionFactory(
  29. 				ActiveMQConnection.DEFAULT_USER,
  30. 				ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
  31. 		try {
  32. 			// 构造从工厂得到连接对象
  33. 			connection = connectionFactory.createConnection();
  34. 			// 启动
  35. 			connection.start();
  36. 			// 获取操作连接
  37. 			session = connection.createSession(Boolean.FALSE,
  38. 					Session.AUTO_ACKNOWLEDGE);
  39. 			// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
  40. 			destination = session.createTopic("FirstTop");
  41. 			consumer = session.createConsumer(destination);
  42. 			// consumer2 = session.createConsumer(destination);
  43. 			System.out.println("订阅者1：消费者程序开始运行...");
  44. 			while (true) {
  45. 				// 设置接收者接收消息的时间，为了便于测试，这里谁定为100s
  46. 				TextMessage message = (TextMessage) consumer.receive(8000);
  47. 				// Object message = (Object) consumer.receive(28000);
  48. 				if (null != message) {
  49. 					System.out.println("订阅者1：收到消息-->" + message.getText());
  50. 				} else {
  51. 					System.out.println("订阅者1：运行结束...
");
  52. 					break;
  53. 				}
  54. 			}
  55. 		} catch (Exception e) {
  56. 			e.printStackTrace();
  57. 		} finally {
  58. 			try {
  59. 				if (null != connection)
  60. 					connection.close();
  61. 			} catch (Throwable ignore) {
  62. 			}
  63. 		}
  64. 	}
  65. }
发布者程序：
  1. package com.ll.activemq;
  3. import javax.jms.Connection;
  4. import javax.jms.ConnectionFactory;
  5. import javax.jms.DeliveryMode;
  6. import javax.jms.Destination;
  7. import javax.jms.MessageProducer;
  8. import javax.jms.Session;
  9. import javax.jms.TextMessage;
  10. import org.apache.activemq.ActiveMQConnection;
  11. import org.apache.activemq.ActiveMQConnectionFactory;
  13. public class Sender {
  14. 	private static final int SEND_NUMBER = 5;
  16. 	public static void main(String[] args) {
  17. 		// ConnectionFactory ：连接工厂，JMS 用它创建连接
  18. 		ConnectionFactory connectionFactory;
  19. 		// Connection ：JMS 客户端到JMS Provider 的连接
  20. 		Connection connection = null;
  21. 		// Session： 一个发送或接收消息的线程
  22. 		Session session;
  23. 		// Destination ：消息的目的地;消息发送给谁.
  24. 		Destination destination;
  25. 		// MessageProducer：消息发送者
  26. 		MessageProducer producer;
  27. 		// TextMessage message;
  28. 		// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
  29. 		connectionFactory = new ActiveMQConnectionFactory(
  30. 				ActiveMQConnection.DEFAULT_USER,
  31. 				ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
  32. 		try {
  33. 			// 构造从工厂得到连接对象
  34. 			connection = connectionFactory.createConnection();
  35. 			// 启动
  36. 			connection.start();
  37. 			// 获取操作连接
  38. 			session = connection.createSession(Boolean.TRUE,
  39. 					Session.AUTO_ACKNOWLEDGE);
  40. 			// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
  41. 			destination = session.createTopic("FirstTop");
  42. 			// 得到消息生成者【发送者】
  43. 			producer = session.createProducer(destination);
  44. 			// 设置不持久化，此处学习，实际根据项目决定
  45. 			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
  46. 			// 构造消息，此处写死，项目就是参数，或者方法获取
  47. 			sendMessage(session, producer);
  48. 			// session.commit();
  49. 			System.out.println("发布者程序退出...");
  50. 		} catch (Exception e) {
  51. 			e.printStackTrace();
  52. 		} finally {
  53. 			try {
  54. 				if (null != connection)
  55. 					connection.close();
  56. 			} catch (Throwable ignore) {
  57. 			}
  58. 		}
  59. 	}
  61. 	public static void sendMessage(Session session, MessageProducer producer)
  62. 			throws Exception {
  63. 		for (int i = 1; i <= SEND_NUMBER; i++) {
  64. 			TextMessage message = session.createTextMessage("ActiveMq 发布的消息"
  65. 					+ i);
  66. 			// 发送消息到目的地方
  67. 			System.out.println("发布消息：" + "ActiveMq 发布的消息" + i);
  68. 			producer.send(message);
  69. 			session.commit();
  70. 			Thread.sleep(10000);
  71. 		}
  72. 	}
  73. }






附件列表

    【推广】 免费学中医，健康全家人
    原文地址：https://www.cnblogs.com/ssslinppp/p/4462195.html