1.版本说明

  本文主要讲解rocketmq的使用。并且这里的rocketmq已经是由阿里移交至apache基金的版本。也就是4.x版本。在本文编写的时候,apache还未提供编译好的rocketmq下载。只能去github下载源码自行编译。本站前面有讲解rocketmq的编译和安装。如果需要了解如何安装rocketmq4.x请点击搜索输入rocket安装,这里不再重述。

2.项目使用环境

1.eclipse
2.maven
3.spring boot

3.demo重点代码

3.1引入rocketmq的依赖包

        <!-- 注意rocketmq 4.x版本客户端是Apache的3.x是阿里的 -->
		<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
		<dependency>
			<groupId>org.apache.rocketmq</groupId>
			<artifactId>rocketmq-client</artifactId>
			<version>4.1.0-incubating</version>
		</dependency>

3.2 rocketmq消息生产者(Producer)

package com.example.simple;

import java.util.concurrent.TimeUnit;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class RocketmqProducer {
	
	public static void main(String[] args) throws MQClientException, InterruptedException, MQClientException {
		/**
		 * 一个应用创建一个Producer,由应用来维护此对象,可以设置为全局对象或者单例<br>
		 * 注意:ProducerGroupName需要由应用来保证唯一<br>
		 * ProducerGroup这个概念发送普通的消息时,作用不大,但是发送分布式事务消息时,比较关键,
		 * 因为服务器会回查这个Group下的任意一个Producer
		 */
		final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
		producer.setNamesrvAddr("192.168.8.202:9876");
		producer.setInstanceName("Producer");
		/**
		 * Producer对象在使用之前必须要调用start初始化,初始化一次即可<br>
		 * 注意:切记不可以在每次发送消息时,都调用start方法
		 */
		producer.start();

		/**
		 * 下面这段代码表明一个Producer对象可以发送多个topic,多个tag的消息。
		 * 注意:send方法是同步调用,只要不抛异常就标识成功。但是发送成功也可会有多种状态,<br>
		 * 例如消息写入Master成功,但是Slave不成功,这种情况消息属于成功,但是对于个别应用如果对消息可靠性要求极高,<br>
		 * 需要对这种情况做处理。另外,消息可能会存在发送失败的情况,失败重试由应用来处理。
		 */
		for (int m = 0; m < 1; m++) {
			for (int i = 0; i < 10; i++) {
				try {
					{
						Message msg = new Message("TopicTest2", // topic
								"push", // tag
								"OrderID00" + i, // key
								("Hello MetaQ" + i).getBytes());// body
						SendResult sendResult = producer.send(msg);
						System.out.print(i);
						System.out.print(sendResult);
						System.out.print(",");
						System.out.print(new String(msg.getBody()));
						System.out.println(",");
					}
					// {
					// Message msg = new Message("TopicTest2",// topic
					// "TagB",// tag
					// "OrderID0034",// key
					// ("Hello MetaQB").getBytes());// body
					// SendResult sendResult = producer.send(msg);
					// System.out.println(sendResult);
					// }
					//
					// {
					// Message msg = new Message("TopicTest3",// topic
					// "TagC",// tag
					// "OrderID061",// key
					// ("Hello MetaQC").getBytes());// body
					// SendResult sendResult = producer.send(msg);
					// System.out.println(sendResult);
					// }
					try {
						TimeUnit.MILLISECONDS.sleep(100);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
				// TimeUnit.MILLISECONDS.sleep(10);
			}
		}
		/**
		 * 应用退出时,要调用shutdown来清理资源,关闭网络连接,从MetaQ服务器上注销自己
		 * 注意:我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法
		 */
		// producer.shutdown();
		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
			public void run() {
				producer.shutdown();
			}
		}));
		System.exit(0);
	}
}

3.3rocketmq消息消费者(Consumer)

package com.example.simple;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

public class RocketmqConsumer {

	/**
	 * 当前例子是PushConsumer用法,使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。<br>
	 * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息,然后再回调用户Listener方法<br>
	 */
	public static void main(String[] args) throws InterruptedException, MQClientException, MQClientException {
		/**
		 * 一个应用创建一个Consumer,由应用来维护此对象,可以设置为全局对象或者单例<br>
		 * 注意:ConsumerGroupName需要由应用来保证唯一
		 */
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
		consumer.setNamesrvAddr("192.168.8.202:9876");
		consumer.setInstanceName("Consumber");
		consumer.setMessageModel(MessageModel.CLUSTERING);
		// consumer.setConsumeThreadMin(1);
		// consumer.setConsumeThreadMax(1);

		/**
		 * 订阅指定topic下tags分别等于TagA或TagC或TagD
		 */
		consumer.subscribe("TopicTest1", "TagA || TagC || TagD");
		/**
		 * 订阅指定topic下所有消息<br>
		 * 注意:一个consumer对象可以订阅多个tag
		 */
		consumer.subscribe("TopicTest2", "*");

		consumer.registerMessageListener(new MessageListenerConcurrently() {

			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

				System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs.size());

				MessageExt msg = msgs.get(0);
				if (msg.getTopic().equals("TopicTest1")) {
					// 执行TopicTest1的消费逻辑
					if (msg.getTags() != null && msg.getTags().equals("TagA")) {
						// 执行TagA的消费
						System.out.println(new String(msg.getBody()));
					} else if (msg.getTags() != null && msg.getTags().equals("TagC")) {
						// 执行TagC的消费
						System.out.println(new String(msg.getBody()));
					} else if (msg.getTags() != null && msg.getTags().equals("TagD")) {
						// 执行TagD的消费
						System.out.println(new String(msg.getBody()));
					}
				} else if (msg.getTopic().equals("TopicTest2")) {
					System.out.println(new String(msg.getBody()));
				}

				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

			}
		});

		/**
		 * Consumer对象在使用之前必须要调用start初始化,初始化一次即可<br>
		 */
		consumer.start();

		System.out.println("ConsumerStarted.");
	}

}

1.main方法运行消息生产者
2.man方法运行消费者
3.观察消费者数据