AI 助理
备案控制台
开发者社区 微服务 文章 正文

RocketMQ-初体验RocketMQ(09)-广播消息、延时消息、批量消息

2024-01-02 77
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： RocketMQ-初体验RocketMQ(09)-广播消息、延时消息、批量消息

广播消息

广播消息概述

广播消息就是向所有用户发送消息。 如果我们希望所有订阅者都能收到有关某个主题的消息，可以使用广播消息。

举个例子 生产者发送10条消息，有2个订阅者，则这两个订阅者会分别收到10条消息， 而与广播模式相对应的集群模式这是 2个订阅者一共收到10条消息。

Rocketmq 消费者默认是集群的方式消费的，使用广播模式进行消费需要显示设置

核心：消费端设置消息模型 consumer.setMessageModel(MessageModel.BROADCASTING);

演示步骤

  • 启动2个或者2个以上的消费者
  • 启动生产者发送消息
  • 观察2个消费者的消息接收情况 ：两个Consumer收到了同样的消息，OK.

生产者：

package com.artisan.rocketmq.broadcast;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
/**
 * @author 小工匠
 * @version v1.0
 * @create 2019-11-10 19:22
 * @motto show me the code ,change the word
 * @blog https://artisan.blog.csdn.net/
 * @description
 **/
public class BroadcastProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("consumer_model_group");
        producer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");
        producer.start();
        for (int i = 0; i < 4; i++){
            Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    ("Hello world"+i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        producer.shutdown();
    }
}

消费者：

package com.artisan.rocketmq.broadcast;
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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import java.util.List;
/**
 * @author 小工匠
 * @version v1.0
 * @create 2019-11-10 19:27
 * @motto show me the code ,change the word
 * @blog https://artisan.blog.csdn.net/
 * @description
 **/
public class BroadcastConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_model_group");
        consumer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //广播,全量消费
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe("TopicTest", "TagA || TagC || TagD");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt ext : msgs){
                    System.out.printf(Thread.currentThread().getName() + " Receive New Message: " + new String(ext.getBody()) + "%n");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Broadcast Consumer Started.%n");
    }
}

测试结果：

生产者：

消费者1：

消费者2：

延时消息

概述

定时消息是指消息发到 Broker 后，不能立刻被 Consumer 消费，要到特定的时间点或者等待特定的时间后才能被消费。

使用场景

举个例子： 电商系统，提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

延时机制

org.apache.rocketmq.store.config.MessageStoreConfig#messageDelayLevel

当前支持的延迟时间

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

分别对应级别

1 2 3....................

设置消息时延

Message message = new Message；
message.setDelayTimeLevel(3)

现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18 消息消费失败会进入延时消息队列，消息发送时间与设置的延时等级和重试次数有关。

实现原理

延迟队列的核心思路: 【利用中间队列临时存储】—>所有的延迟消息由producer消息发憷之后，都会存放在一个topic下 （SHCEDULE_TOPIC_XXXX）, 不同的延迟级别对应不同的队列序号，当延迟时间到了之后，由定时线程读取转换为普通的消息存到真实指定的topic下，此时对于consumer端此消息才可见，从而被consumer消费。

示例

生产者：

package com.artisan.rocketmq.schedule;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import java.util.Date;
/**
 * @author 小工匠
 * @version v1.0
 * @create 2019-11-10 17:23
 * @motto show me the code ,change the word
 * @blog https://artisan.blog.csdn.net/
 * @description
 **/
public class ScheduledMessageProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ExampleConsumer");
        producer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");
        producer.start();
        int totalMessagesToSend = 3;
        for (int i = 0; i < totalMessagesToSend; i++) {
            Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
            //延时消费  6-->2分钟
            message.setDelayTimeLevel(6);
            // Send the message
            producer.send(message);
        }
        System.out.printf("message send is completed .%n" + new Date());
        producer.shutdown();
    }
}

消费者：

package com.artisan.rocketmq.schedule;
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.common.message.MessageExt;
import java.util.Date;
import java.util.List;
/**
 * @author 小工匠
 * @version v1.0
 * @create 2019-11-10 17:23
 * @motto show me the code ,change the word
 * @blog https://artisan.blog.csdn.net/
 * @description
 **/
public class ScheduledMessageConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer");
        consumer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");
        consumer.subscribe("TestTopic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
                for (MessageExt message : messages) {
                    // Print approximate delay time period
                    System.out.println(new Date() + "Receive message[msgId=" + message.getMsgId() + "] "
                            + "message content is :" + new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        //System.out.printf("Consumer Started.%n");
    }
}

设置的延迟level为6 ，对应的时间间隔是两分钟，OK。

批量消息

批量消息概述

批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，而且不能是延时消息

此外，这一批消息的总大小不应超过4MB。rocketmq建议每次批量消息大小大概在1MB。当消息大小超过4MB时，需要将消息进行分割

示例

生产者

package com.artisan.rocketmq.batch;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import java.util.ArrayList;
import java.util.List;
/**
 * @author 小工匠
 * @version v1.0
 * @create 2019-11-10 21:27
 * @motto show me the code ,change the word
 * @blog https://artisan.blog.csdn.net/
 * @description
 **/
public class BatchProducer {
    public static void main(String[] args) throws Exception {
        /**
         * rocketMq 支持消息批量发送
         * 同一批次的消息应具有：相同的主题，相同的waitStoreMsgOK，并且不支持定时任务。
         * <strong> 同一批次消息建议大小不超过~1M </strong>,消息最大不能超过4M,需要
         * 对msg进行拆分
         */
        DefaultMQProducer producer = new DefaultMQProducer("batch_group");
        producer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");
        producer.start();
        String topic = "BatchTest";
        List<Message> messages = new ArrayList<>();
        messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
        ListSplitter splitter = new ListSplitter(messages);
        /**
         * 对批量消息进行拆分
         */
        while (splitter.hasNext()) {
            try {
                List<Message>  listItem = splitter.next();
                producer.send(listItem);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        producer.shutdown();
    }
}

消息拆分

package com.artisan.rocketmq.batch;
import org.apache.rocketmq.common.message.Message;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
/**
 * @author 小工匠
 * @version v1.0
 * @create 2019-11-10 21:35
 * @motto show me the code ,change the word
 * @blog https://artisan.blog.csdn.net/
 * @description
 **/
public class ListSplitter implements Iterator<List<Message>> {
    private final int SIZE_LIMIT = 1000 * 1000 * 1;//1MB
    private final List<Message> messages;
    private int currIndex;
    public ListSplitter(List<Message> messages) {
        this.messages = messages;
    }
    @Override
    public boolean hasNext() {
        return currIndex < messages.size();
    }
    @Override
    public List<Message> next() {
        int nextIndex = currIndex;
        int totalSize = 0;
        //遍历消息准备拆分
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; //for log overhead
            if (tmpSize > SIZE_LIMIT) {
                if (nextIndex - currIndex == 0) {
                    nextIndex++;
                }
                break;
            }
            if (tmpSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
        }
        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }
}

消费者

package com.artisan.rocketmq.batch;
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 java.util.List;
/**
 * @author 小工匠
 * @version v1.0
 * @create 2019-11-10 21:38
 * @motto show me the code ,change the word
 * @blog https://artisan.blog.csdn.net/
 * @description
 **/
public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch_group");
        consumer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");
        consumer.subscribe("BatchTest", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs){
                    System.out.println("queueId=" + msg.getQueueId() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

代码

请移步： https://github.com/yangshangwei/rocketmqMaster


文章标签：
云消息队列 MQ
消息中间件
RocketMQ
存储
关键词：
云消息队列 MQ延时消息
云消息队列 MQ延时
云消息队列 MQ初体验
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
小小工匠
目录
相关文章
小小工匠
|
7月前
|
消息中间件 Apache 开发工具
RocketMQ-初体验RocketMQ(08)-IDEA拉取调测RocketMQ源码
RocketMQ-初体验RocketMQ(08)-IDEA拉取调测RocketMQ源码
小小工匠
73 0
小小工匠
|
7月前
|
消息中间件 存储 JSON
RocketMQ-初体验RocketMQ(05)_RocketMQ架构解读
RocketMQ-初体验RocketMQ(05)_RocketMQ架构解读
小小工匠
78 0
栈江湖
|
11天前
|
消息中间件 中间件 Kafka
MQ四兄弟：如何实现延时消息
本文介绍了几种常见的消息队列系统（RabbitMQ、RocketMQ、Kafka和Pulsar）实现延时消息的方式。RabbitMQ通过死信队列或延时插件实现；RocketMQ内置延时消息支持，可通过设置`delayTimeLevel`属性实现；Kafka不直接支持延时消息，但可以通过时间戳、延时Topic、Kafka Streams等方法间接实现；Pulsar自带延时消息功能，提供`deliverAfter`和`deliverAt`两种方式。每种方案各有优劣，适用于不同的应用场景。
栈江湖
39 0
三分钟热度的鱼
|
6月前
|
消息中间件 RocketMQ
消息队列 MQ产品使用合集之在开源延时消息插件方案中和原生延时消息方案中，同时设置参数是否会出现错乱
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
65 1
小小工匠
|
7月前
|
消息中间件 SQL RocketMQ
RocketMQ-初体验RocketMQ(10)-过滤消息_SQL92表达式筛选消息
RocketMQ-初体验RocketMQ(10)-过滤消息_SQL92表达式筛选消息
小小工匠
159 0
小小工匠
|
7月前
|
消息中间件 监控 Shell
RocketMQ-初体验RocketMQ(03)_RocketMQ多机集群部署
RocketMQ-初体验RocketMQ(03)_RocketMQ多机集群部署
小小工匠
91 0
小小工匠
|
7月前
|
消息中间件 Java RocketMQ
RocketMQ-初体验RocketMQ(11)-过滤消息_自定义Java类筛选消息
RocketMQ-初体验RocketMQ(11)-过滤消息_自定义Java类筛选消息
小小工匠
70 0
RocketMQ-初体验RocketMQ(11)-过滤消息_自定义Java类筛选消息
小小工匠
|
7月前
|
消息中间件 API RocketMQ
RocketMQ-初体验RocketMQ(07)-使用API操作RocketMQ_顺序消息 ordermessage
RocketMQ-初体验RocketMQ(07)-使用API操作RocketMQ_顺序消息 ordermessage
小小工匠
65 0
小小工匠
|
7月前
|
消息中间件 存储 Java
RocketMQ-初体验RocketMQ(06)-使用API操作RocketMQ ,理解RocketMQ的存储结构
RocketMQ-初体验RocketMQ(06)-使用API操作RocketMQ ,理解RocketMQ的存储结构
小小工匠
206 0
勇哥Java实战
|
消息中间件 中间件 数据库
RocketMQ 事务消息初体验
事务消息是 RocketMQ 的高级特性之一 。这篇文章，笔者会从应用场景、功能原理、实战例子三个模块慢慢为你揭开事务消息的神秘面纱。
勇哥Java实战
5369 8

热门文章

最新文章

  • 1
    大厂都在用的分布式事务方案，Seata+RocketMQ带你打破10万QPS瓶颈
  • 2
    C5GAME 游戏饰品交易平台借助 RocketMQ Serverless 保障千万级玩家流畅体验
  • 3
    朗新科技集团如何用云消息队列 RocketMQ 版“快、准、狠”破解业务难题？
  • 4
    2024最全RabbitMQ集群方案汇总
  • 5
    【实践】快速学会使用云消息队列RabbitMQ版
  • 6
    说说MQ在你项目中的应用(一)
  • 7
    高弹性、低成本的云消息队列RabbitMQ 版
  • 8
    2024最全RocketMQ集群方案汇总
  • 9
    如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
  • 10
    多协议网关BL110钡铼6路RS485转MQTT协议云网关
  • 1
    RabbitMQ的springboot项目集成使用-01
    150
  • 2
    深入理解MQ消息队列的高可用与可靠性策略
    1352
  • 3
    rabbitmq插件升级-延时队列为列
    670
  • 4
    rabbitmq安装erlang环境后没生效
    1047
  • 5
    Springboot 集成Rabbitmq之延时队列
    196
  • 6
    RabbitMQ高级
    50
  • 7
    rabbitMQ集群
    60
  • 8
    涉及rocketMQ，jemeter等性能测试服务器的安装记录
    82
  • 9
    用docker方式 安装rabbitmq 并配置MQTT
    336
  • 10
    MQTT 5.0 报文解析 02：PUBLISH 与 PUBACK
    451

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 通过轻量消息队列（原MNS）主题HTTP订阅+ARMS实现自定义数据多渠道告警
  • 快速体验云消息队列RocketMQ版
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 快速入门轻量消息队列（原MNS）
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：定时消息通知功能体验
    • 下一篇
    手把手教你白嫖阿里云服务器(免费领服务器)