ACtiveMQ中间件-发布订阅模式

前言：ActiveMQ学习心得

1.MQ是什么

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。其中较为成熟的MQ产品有IBM WEBSPHERE MQ、Apache ActiveMQ等等。

1.1.ActiveMQ的概述

ActiveMQ是一种开源的，实现了JMS1.1规范的，面向消息(MOM)的中间件，为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。

1.3.ActiveMQ的实现原理



如图所示：当客户端A和客户端B通信，客户端A发送消息到队列或主题中，然后等待客户端B接收消息。当客户端B接收消息并且确认消息后，消息才会

从队列中删除。

2.ActiveMQ的下载

2.1.下载路径

官方网站下载：http://activemq.apache.org/

2.2ActiveMQ的运行

我下载的是active5.9版本的进入ActiveMQ的解压包bin目录开始启动服务



2.2.ActiveMQ发布订阅模式代码示例

首先添加jar包为



开始编写代码既然是消息中间件就是来处理消息的，那么处理消息就有发送方和接收方，在发布订阅模式下面需先运行消息接收方在运行发送方可接收到消息

发送方代码

/**

 * @摘要 测试发送单条数据的类

 */

public class ZMQOneSendTest{

	public static void main(String[] args) throws Exception {

		// 4.开始建立连接时间

		Long nStartTime = System.currentTimeMillis();

		// 5.建立连接工厂

		org.apache.activemq.ActiveMQConnectionFactory nFactory = new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616");

		// 6.用工厂建立Connection连接

		javax.jms.TopicConnection nConnection = nFactory.createTopicConnection();

		// 7.创建会话Session，不启动事务，签收模式为自动确认模式

		javax.jms.TopicSession nSession = nConnection.createTopicSession(false,javax.jms.Session.AUTO_ACKNOWLEDGE);

		// 8.创建主题

		javax.jms.Topic nTopic = nSession.createTopic("199");

		// 9.创建发布者

		javax.jms.TopicPublisher nPublihser = nSession.createPublisher(nTopic);

		// 10.设置持久化

		nPublihser.setDeliveryMode(javax.jms.DeliveryMode.PERSISTENT);

		// 11.启动连接

		nConnection.start();

		// 12.查询开始时间

		Long nStartSelectTime = System.currentTimeMillis();

		// 14.查询结束时间

		Long nEndSelectTime = System.currentTimeMillis();

		// 15.创建消息

		javax.jms.TextMessage nMessage = nSession.createTextMessage();

		nMessage.setText("我是发送的数据");

		// 16.发送开始时间

		Long nStartSendTime = System.currentTimeMillis();

		// 17.发布者发布消息

		nPublihser.publish(nMessage);

		// 18.发布结束时间

	    Long nEndSendTime = System.currentTimeMillis();

		// 19.打印消息

	    System.out.println("已发送消息 "+nMessage);

	    // 关闭发布者、会话、连接

	    nPublihser.close();

	    nSession.close();

	    nConnection.close();

	    // 20.计算时间

		System.out.println("查询消息时间 "+(nEndSelectTime-nStartSelectTime));

		System.out.println("发送消息时间"+(nEndSendTime-nStartSendTime));

		System.out.println("总时间"+(nEndSendTime-nStartTime));

	}

}

接收方代码

import javax.jms.JMSException;

public class ZMQReadTest {

	public static void main(String[] args) throws JMSException {

		// 1.建立连接工厂

		org.apache.activemq.ActiveMQConnectionFactory nFactory = new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616");

		// 2.用工厂建立Connection连接

		javax.jms.TopicConnection nTopicConnection = nFactory.createTopicConnection();

		// 3.为客户端设置一个唯一ID

		nTopicConnection.setClientID("5");

		// 4.创建会话Session，不启动事务，签收模式为自动确认模式

		javax.jms.TopicSession nSession = nTopicConnection.createTopicSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);

		// 5.创建主题

		javax.jms.Topic nTopic = nSession.createTopic("199");

		// 6.创建订阅者

		javax.jms.TopicSubscriber nTopicSubscriber = nSession.createDurableSubscriber(nTopic, "min");

		// 6.1.为消费者添加消息监听器，onMessage方法接收消息

		nTopicSubscriber.setMessageListener(new javax.jms.MessageListener() {

			public void onMessage(javax.jms.Message nMessage) {

				try {

					if (nMessage instanceof javax.jms.TextMessage) {

						String nTextMessage = ((javax.jms.TextMessage) nMessage).getText();

						System.out.println("收到的消息：" + nTextMessage);

					}

				} catch (javax.jms.JMSException e) {

					e.printStackTrace();

				}

			}

		});

		// 7.开启连接

		nTopicConnection.start();

	}

}

测试消息的发送和接收是否成功时需要先运行消费者(消息的接收者)再运行生产者(消息发送者)当显示如下消息接收方和发送方打印消息如下即可。











好的给大家写好一个发布订阅案例了不懂的可以问我哈！！！