RocketMQ 相关知识

Bertha 。 2022-06-09 10:21 172阅读 0赞

# RocketMQ 相关知识 #

> Rocket MQ消息队列（Message Queue，简称 MQ）是阿里巴巴集团中间件技术部自主研发的专业消息中间件。产品基于高可用分布式集群技术，提供消息发布订阅、消息轨迹查询、定时（延时）消息、资源统计、监控报警等一系列消息云服务，是企业级互联网架构的核心产品。

## Rocket MQ相关名词 ##

*  Producer 消息生产者，负责生产消息
 *  Consumer 消息消费者，负责消费消息
 *  NameServer 无状态节点，用来保存活跃的broker列表和topic列表
 *  Broker 消息中转角色，负责存储消息，转发消息
 *  Topic 消息的逻辑管理单位
 *  Message 消息  
    
    
     *  body 消息体，用于携带消息具体内容
     *  key 消息的key，用于区别不同的消息
     *  tags 消息的Tag，用于不同的订阅者过滤消息

## 消息发送方式 ##

*  同步方式  
    
    
    > 发送消息，接收到结果之后再发送下一条消息，速度最慢，耗时最长
 *  异步方式  
    发送消息，不论是否收到结果，直接发送下一条消息，发送速度介于同步和单向方式之间
 *  单向方式  
    发送消息，直接发送消息，不返回发送结果，发送速度最快

## 消息类型 ##

*  定时消息  
    
    
    > 在指定的发送时间发送消息
 *  延时消息  
    从当前时间开始，经过延时时间后再发送消息
 *  顺序消息  
    立即发送消息
 *  事务消息  
    MQ 提供类似 X/Open XA 的分布事务功能，通过 MQ 事务消息能达到分布式事务的最终一致

# 示例代码 #

## Producer ##

public class ProducerDelayTest { 
        public static void main(String[] args) {
            Properties properties = new Properties();
            //您在 MQ 控制台创建的Producer ID
            properties.put(PropertyKeyConst.ProducerId, "XXX");
            // 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey, "XXX");
            // 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, "XXX");
            // 设置 TCP 接入域名（此处以公共云生产环境为例）
            properties.put(PropertyKeyConst.ONSAddr,
              "http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal");
            Producer producer = ONSFactory.createProducer(properties);
            // 在发送消息前，必须调用start方法来启动Producer，只需调用一次即可。
            producer.start();
    
                /** * 消息类型代码，参考下面消息类型代码 */
                 /** * 消息发送方式代码，参考下面发送方式代码 */
            System.out.println("Message Id:" + sendResult.getMessageId());
            // 在应用退出前，销毁 Producer 对象
            // 注意：如果不销毁也没有问题，如果发送消息较多不应该销毁
            producer.shutdown();
        }
    }

### 消息类型代码 ###

*  定时消息

Message msg = new Message();
            msg.setTag("TAG");
            msg.setKey("KEY");
            msg.setTopic("TOPIC");
            msg.setBody("BODY".getBytes());
            long timeStamp =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2017-09-03 16:21:00").getTime();
            msg.setStartDeliverTime(timeStamp);

*  延时消息

Message msg = new Message();
            msg.setTag("TAG");
            msg.setKey("KEY");
            msg.setTopic("TOPIC");
            msg.setBody("BODY".getBytes()); 
            long delayTime = 3000;//30秒后再发送
            msg.setStartDeliverTime(System.currentTimes() + delayTime);

*  顺序消息

Message msg = new Message();
            msg.setTag("TAG");
            msg.setKey("KEY");
            msg.setTopic("TOPIC".getBytes());
            msg.setBody("BODY");

### 消息发送方式代码 ###

*  同步方式发送

SendResult sendResult = producer.send(msg);

*  异步方式发送

producer.sendAsync(message, new SendCallback() {
                @Override
                public void onSuccess(final SendResult sendResult) {
                    logger.info("MQ send ASYNCHRONOUS message successed，response is " + JSON.toJSONString(sendResult));
                }
    
                @Override
                public void onException(OnExceptionContext onExceptionContext) {
                    logger.info("MQ send ASYNCHRONOUS message failed, error is " + onExceptionContext.getException().getMessage());
                }
            });

*  单向方式发送

producer.sendOneway(message);

## Consumer ##

public class ConsumerTest {
        public static void main(String[] args) {
            Properties properties = new Properties();
            // 您在控制台创建的 Consumer ID
            properties.put(PropertyKeyConst.ConsumerId, "XXX");
            // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey, "XXX");
            // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, "XXX");
            // 设置 TCP 接入域名（此处以公共云生产环境为例）
            properties.put(PropertyKeyConst.ONSAddr,
              "http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal");
              // 集群订阅方式 (默认)
              // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
              // 广播订阅方式
              // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
            Consumer consumer = ONSFactory.createConsumer(properties);
            consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() { //订阅多个Tag
                public Action consume(Message message, ConsumeContext context) {
                    System.out.println("Receive: " + message);
                    return Action.CommitMessage;
                }
            });
            //订阅另外一个Topic
            consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() { //订阅全部Tag
                public Action consume(Message message, ConsumeContext context) {
                    System.out.println("Receive: " + message);
                    return Action.CommitMessage;
                }
            });
            consumer.start();
            System.out.println("Consumer Started");
        }
    }

--------------------

*  Spring Boot Demo 参照[https://github.com/helloworlde/SpringBoot-RocketMQ][https_github.com_helloworlde_SpringBoot-RocketMQ]

[https_github.com_helloworlde_SpringBoot-RocketMQ]: https://github.com/helloworlde/SpringBoot-RocketMQ