阿里云消息队列RocketMQ使用示例

本文代码示例参见：https://gitee.com/imlichao/RocketMQ-example

Apache RocketMQ文档：http://rocketmq.apache.org/docs/quick-start/

阿里云RocketMQ文档：https://help.aliyun.com/product/29530.html

简介

消息队列 RocketMQ 是阿里巴巴集团自主研发的专业消息中间件，基于高可用分布式集群技术，提供消息订阅和发布、消息轨迹查询以及定时（延时）消息、资源统计、监控报警等一系列消息云服务，是企业级互联网架构的核心产品。 消息队列 RocketMQ 历史超过9年，为分布式应用系统提供异步解耦、削峰填谷的能力，同时具有海量消息堆积、高吞吐、可靠重试等互联网应用所需的特性，是阿里巴巴双11使用的核心产品。

消息队列 RocketMQ 是阿里云正式商用的产品，目前在阿里云多个地域（Region）提供了高可用消息云服务，单个域内采用多机房部署，可用性极高，即便整个机房都不可用，仍然能够为应用提供消息发布服务，产品稳定性及可用性彻底按照阿里巴巴内部标准来实施，无单点。

消息队列 RocketMQ 目前提供 TCP 和 HTTP 协议层面的接入方式，支持 Java、C++、 .NET、Go、Python、Nodejs、PHP 这七种编程语言，方便不一样编程语言开发的应用快速接入消息队列 RocketMQ 消息云服务。 用户能够将应用部署在阿里云 ECS、企业自建云，或者嵌入到移动端、物联网设备中与消息队列 RocketMQ 创建链接进行消息收发，同时本地开发者也能够经过公网接入消息队列 RocketMQ 服务进行消息收发。

 

消息收发模型

消息队列 RocketMQ 支持“发布/订阅”模型，消息发布者（生产者）能够将一条消息发送服务端的某个主题（Topic），多个消息接收方（消费者）订阅这个主题以接收该消息，以下图所示：

 

示例

本例使用阿里云RocketMQ产品，其中用户名密码地址等使用“XXXXXX”表示。

增长maven依赖

<!-- 增长RocketMQ依赖 -->
<dependency>
  <groupId>com.aliyun.openservices</groupId>
  <artifactId>ons-client</artifactId>
  <version>1.8.0.Final</version>
</dependency>

配置类

本例使用两个同组下的消费者共同消费，而且消息采用exactly-once语义保证每一个消息只被消费一次

package pub.imlichao.config;

import com.aliyun.openservices.ons.api.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;


/**
 * RocketMQ配置
 */
@Configuration
public class RocketMQConfig {
	/**
	 * 生产者配置
	 * @return
	 */
	@Bean
	public Producer  producer () {
		Properties properties = new Properties();
		// 鉴权用 AccessKey，在阿里云服务器管理控制台建立
		properties.put(PropertyKeyConst.AccessKey,"XXXXXX");
		// 鉴权用 SecretKey，在阿里云服务器管理控制台建立
		properties.put(PropertyKeyConst.SecretKey, "XXXXXX");
		// 设置 TCP 接入域名，进入控制台的实例管理页面，在页面上方选择实例后，在实例信息中的“获取接入点信息”区域查看
		properties.put(PropertyKeyConst.NAMESRV_ADDR,"XXXXXX");
		//经过 PropertyKeyConst.EXACTLYONCE_DELIVERY 开启 exactly-once 投递语义(保证拥有多个消费者时消息只被消费一次)
		properties.put(PropertyKeyConst.EXACTLYONCE_DELIVERY, "true");
		Producer producer = ONSFactory.createProducer(properties);
		// 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次便可
		producer.start();

		return producer;
	}

	/**
	 * 消费者1配置
	 * @return
	 */
	@Bean
	public Consumer consumer1 () {
		Properties properties = new Properties();
		// 您在控制台建立的 Group ID
		properties.put(PropertyKeyConst.GROUP_ID, "GID_pmall_consumer");
		// 鉴权用 AccessKey，在阿里云服务器管理控制台建立
		properties.put(PropertyKeyConst.AccessKey,"XXXXXX");
		// 鉴权用 SecretKey，在阿里云服务器管理控制台建立
		properties.put(PropertyKeyConst.SecretKey, "XXXXXX");
		// 设置 TCP 接入域名，进入控制台的实例管理页面，在页面上方选择实例后，在实例信息中的“获取接入点信息”区域查看
		properties.put(PropertyKeyConst.NAMESRV_ADDR,"XXXXXX");
		Consumer consumer = ONSFactory.createConsumer(properties);
		//建立消息监听和消息处理逻辑
		consumer.subscribe("pmall_message", "data_storage", new MessageListener() {
			@Override
			public Action consume(Message message, ConsumeContext context) {
				System.out.println("Receive1: " +  new String (message.getBody()) + " " + message.getMsgID());
				return Action.CommitMessage;
			}
		});
		//启动监听
		consumer.start();

		return consumer;
	}

	/**
	 * 消费者2配置
	 * @return
	 */
	@Bean
	public Consumer consumer2 () {
		Properties properties = new Properties();
		// 您在控制台建立的 Group ID
		properties.put(PropertyKeyConst.GROUP_ID, "GID_pmall_consumer");
		// 鉴权用 AccessKey，在阿里云服务器管理控制台建立
		properties.put(PropertyKeyConst.AccessKey,"XXXXXX");
		// 鉴权用 SecretKey，在阿里云服务器管理控制台建立
		properties.put(PropertyKeyConst.SecretKey, "XXXXXX");
		// 设置 TCP 接入域名，进入控制台的实例管理页面，在页面上方选择实例后，在实例信息中的“获取接入点信息”区域查看
		properties.put(PropertyKeyConst.NAMESRV_ADDR,"XXXXXX");
		Consumer consumer = ONSFactory.createConsumer(properties);
		//建立消息监听和消息处理逻辑
		consumer.subscribe("pmall_message", "data_storage", new MessageListener() {
			@Override
			public Action consume(Message message, ConsumeContext context) {
				System.out.println("Receive2: " + new String (message.getBody()) + " " + message.getMsgID());
				return Action.CommitMessage;
			}
		});
		//启动监听
		consumer.start();

		return consumer;
	}
}

发送消息

package pub.imlichao;

import com.aliyun.openservices.ons.api.*;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import pub.imlichao.config.RocketMQConfig;

import javax.annotation.Resource;
import java.util.Date;

@Controller
public class ProducerTest {
    @Resource
    private RocketMQConfig rocketMQConfig;

    /**
     * 向RocketMQ发送消息
     * @return
     */
    @GetMapping(value = "/send")
    public String send(){
        //循环发送消息100次
        for (int i =0;i<100;i++){
            //建立消息
            Message msg = new Message(
                    // 在控制台建立的 Topic，即该消息所属的 Topic 名称
                    "pmall_message",
                    // Message Tag。可理解为 Gmail 中的标签，对消息进行再归类，方便 Consumer 指定过滤条件在消息队列 RocketMQ 服务器过滤
                    "data_storage",
                    // Message Body。任何二进制形式的数据，消息队列 RocketMQ 不作任何干预。须要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                    ("pmall MQ "+ new Date()).getBytes());

            // 设置表明消息的业务关键属性，请尽量全局惟一，以方便您在没法正常收到消息状况下，可经过控制台查询消息并补发。注意：不设置也不会影响消息正常收发
            msg.setKey("data_id");
            // 发送消息，只要不抛异常就是成功
            SendResult sendResult = rocketMQConfig.producer().send(msg);
            //打印 Message ID，以便用于消息发送状态查询
            System.out.println("Send Message success. Message ID is: " + sendResult.getMessageId());
        }
        return "redirect:/";
    }
}