四、RocketMQ发送普通消息、批量消息和延迟消息

Producer发送普通消息的方式

1.同步发送消息

同步消息代表发送端发送消息到broker之后,等待消息发送结果后,再次发送消息
在这里插入图片描述

实现步骤

  1. 创建生产端,声明在哪个生产组
  2. 注册NameServer地址
  3. 构建Message实体,指定topic、tag、body
  4. 启动生产端
  5. 发送消息
@Test
public void syncSend() throws MQBrokerException, RemotingException, InterruptedException, MQClientException {// 1.创建生产端,声明在哪个生产组DefaultMQProducer producer = new DefaultMQProducer("test_group");// 2.注册NameServer地址producer.setNamesrvAddr(NAME_SERVER_ADDR);// 3.构建Message实体,指定topic、tag、bodyMessage message = new Message("test", "hello world".getBytes());// 4.启动生产端producer.start();// 5.发送消息SendResult sendResult = producer.send(message);System.out.println(sendResult.getSendStatus());
}

2.异步发送消息

异步消息代表发送端发送完消息后,会直接返回,但是可以注册一个回调函数,当broker将消息落盘后,回调这个回调函数
在这里插入图片描述

实现步骤

  1. 创建生产端,声明在哪个生产组
  2. 注册NameServer地址
  3. 构建Message实体,指定topic、tag、body
  4. 启动生产端
  5. 发送消息,并且实现SendCallback接口

注:这里必须等待异步返回,否则消费者无法消费成功

@Test
public void asyncSend() throws  RemotingException, InterruptedException, MQClientException {DefaultMQProducer producer = new DefaultMQProducer("test_group");producer.setNamesrvAddr(NAME_SERVER_ADDR);Message message = new Message("test", "tag-a","hello world".getBytes());producer.start();CountDownLatch countDownLatch = new CountDownLatch(1);// 发送消息,并且实现SendCallback接口producer.send(message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {countDownLatch.countDown();System.out.println("发送成功:" + sendResult.getSendStatus());}@Overridepublic void onException(Throwable e) {countDownLatch.countDown();System.out.println("发送失败:" + e);}});countDownLatch.await();
}

3、发送单向消息

发送方只负责发送消息,不等待服务端返回响应且没有回调函数触发,即只发送请求不等待应答。此方式发送消息的过程耗时非常短

在这里插入图片描述

实现步骤

  1. 创建生产端,声明在哪个生产组
  2. 注册NameServer地址
  3. 构建Message实体,指定topic、tag、body
  4. 启动生产端
  5. 发送单向消息
@Test
public void sendOneWay() throws  RemotingException, InterruptedException, MQClientException {DefaultMQProducer producer = new DefaultMQProducer("test_group");producer.setNamesrvAddr(NAME_SERVER_ADDR);Message message = new Message("test","tag-a", "hello world".getBytes());producer.start();producer.sendOneway(message);
}

Producer发送批量消息

在对吞吐率有一定要求的情况下,Apache RocketMQ可以将一些消息聚成一批以后进行发送,可以增加吞吐率,并减少API和网络调用次数。

@Test
public void sendBatch() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER_ADDR);// 构造批量消息List<Message> list = new ArrayList<>();list.add(new Message(RocketMQConfig.TEST_TOPIC, "hello world0".getBytes(Charset.defaultCharset())));list.add(new Message(RocketMQConfig.TEST_TOPIC, "hello world1".getBytes(Charset.defaultCharset())));list.add(new Message(RocketMQConfig.TEST_TOPIC, "hello world2".getBytes(Charset.defaultCharset())));producer.start();// 发送批量消息producer.send(list);producer.shutdown();
}

**注:**需要注意的是批量消息的大小不能超过 1MiB(否则需要自行分割),其次同一批 batch 中 topic 必须相同。

Producer发送延迟消息

Producer想要发送延迟消息,只要设置Message的DelayTimeLevel属性大于0即可。

RocketMQ无法随意设置延迟消息的延迟时间,只能根据延迟级别进行

延迟级别和延迟时间的对应关系

延迟级别延迟时间延迟级别延迟时间
11s106min
25s117min
310s128min
430s139min
51min1410min
62min1520min
73min1630min
84min171h
95min182h
@Test
public void sendDelay() throws Exception {DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER_ADDR);producer.start();Message message = new Message(RocketMQConfig.TEST_TOPIC, "hello world".getBytes(Charset.defaultCharset()));// 设置延迟级别message.setDelayTimeLevel(3);// 发送批量消息SendResult sendResult = producer.send(message);System.out.println(sendResult.getSendStatus());producer.shutdown();
}

延迟消息的原理

延迟消息并不会直接发送到指定的topic,而是发送到一个延迟消息对应的topic中

当延迟消息的时间到达后,在将消息发送到指定的topic中

延迟消息投递的流程

  1. producer端设置消息delayLevel延迟级别,消息属性DELAY中存储了对应了延时级别

  2. broker端收到消息后,判断延时消息延迟级别,如果大于0,则备份消息原始topic,queueId,并将消息topic改为延时消息队列特定topic(SCHEDULE_TOPIC),queueId改为延时级别的delayLevel-1

  3. mq服务端ScheduleMessageService中,为每一个延迟级别单独设置一个定时器,定时(每隔1秒)拉取对应延迟级别的消费队列

  4. 根据消费偏移量offset从commitLog中解析出对应消息

  5. 从消息tagsCode中解析出消息应当被投递的时间,与当前时间做比较,判断是否应该进行投递

  6. 若到达了投递时间,则构建一个新的消息,并从消息属性中恢复出原始的topic,queueId,并清除消息延迟属性,从新进行消息投递

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/104412.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

GBJ2510-ASEMI电源控制柜专用GBJ2510

编辑&#xff1a;ll GBJ2510-ASEMI电源控制柜专用GBJ2510 型号&#xff1a;GBJ2510 品牌&#xff1a;ASEMI 封装&#xff1a;GBJ-4 恢复时间&#xff1a;&#xff1e;50ns 正向电流&#xff1a;25A 反向耐压&#xff1a;1000V 芯片个数&#xff1a;4 引脚数量&#xf…

HBase 表如何按照某表字段排序后顺序存储的方法?

首先需要明白HBase表的排序规则&#xff1a; &#xff08;1&#xff09;rowkey排序&#xff08;字典排序&#xff09;——升序 &#xff08;2&#xff09;Column排序&#xff08;字典排序&#xff09;——升序 &#xff08;3&#xff09;时间戳排序——降序 rowkey 字典序排序…

计算机毕业设计选什么题目好?springboot 医院门诊在线预约挂号系统

✍✍计算机编程指导师 ⭐⭐个人介绍&#xff1a;自己非常喜欢研究技术问题&#xff01;专业做Java、Python、微信小程序、安卓、大数据、爬虫、Golang、大屏等实战项目。 ⛽⛽实战项目&#xff1a;有源码或者技术上的问题欢迎在评论区一起讨论交流&#xff01; ⚡⚡ Java实战 |…

高数笔记03:几何、物理应用

图源&#xff1a;文心一言 本文是我学习高等数学几何、物理应用的一些笔记和心得&#xff0c;希望可以与考研路上的小伙伴一起努力上岸~~&#x1f95d;&#x1f95d; 第1版&#xff1a;查资料、画导图~&#x1f9e9;&#x1f9e9; 参考资料&#xff1a;《高等数学 基础篇》武…

PyTorch 深度学习之多分类问题Softmax Classifier(八)

1. Revision: Diabetes dataset 2. Design 10 outputs using Sigmoid? 2.1 Output a Distribution of prediction with Softmax 2.2 Softmax Layer Example, 2.3 Loss Function-Cross Entropy Cross Entropy in Numpy Cross Entropy in PyTorch 注意交叉熵损失&#xff0c;最…

Vscode中使用Romote远程开发调试Ros2环境

首先&#xff0c;成功安装ros2环境&#xff0c;参考官方文档中的教程&#xff0c;能用运行出来此处的代码 Writing a simple publisher and subscriber (Python) — ROS 2 Documentation: Iron documentation 下载vscode&#xff0c;进行远程开发&#xff0c;具体参考&#xf…

消息队列缓存,以蓝牙消息服务为例

前言 消息队列缓存&#xff0c;支持阻塞、非阻塞模式&#xff1b;支持协议、非协议模式 可自定义消息结构体数据内容 使用者只需设置一些宏定义、调用相应接口即可 这里我用蓝牙消息服务举例 有纰漏请指出&#xff0c;转载请说明。 学习交流请发邮件 1280253714qq.com 原…

IDEA通过Docker插件部署SpringBoot项目

1、配置Docker远程连接端口 找到并编辑服务器上的docker.service文件。 vim /usr/lib/systemd/system/docker.service在下面ExecStart替换成下面的 ExecStart/usr/bin/dockerd -H tcp://0.0.0.0:2375 -H unix://var/run/docker.sock2.重启docker systemctl daemon-reload s…

Kafka 开启SASL/SCRAM认证 及 ACL授权(三)验证

Kafka 开启SASL/SCRAM认证 及 ACL授权(三)验证。 官网地址:https://kafka.apache.org/ 本文说明如何做client验证ACL是否生效,我们之前开启了无acl信息不允许访问的配置。涉及的client有以下几个场景:shell脚本、python脚本、java应用、flink流。 kafka shell script验证…

墨者学院 WordPress 远程命令执行漏洞(CVE-2018-15877)

1. 背景介绍 近日&#xff0c;WordPress 插件Plainview Activity Monitor被曝出存在一个远程命令执行漏洞。Plainview Activity Monitor 是一款网站用户活动监控插件。 远程攻击者可以通过构造的url来诱导wordpress管理员来点击恶意链接最终导致远程命令执行 2.影响范围 Pla…

07_03文件系统怎么玩的

文件系统 Linux将文件系统分为了两层&#xff1a;VFS&#xff08;虚拟文件系统&#xff09;、具体文件系统&#xff0c;如下图所示&#xff1a; VFS&#xff08;Virtual Filesystem Switch&#xff09;称为虚拟文件系统或虚拟文件系统转换&#xff0c;是一个内核软件层&#…

干货:如何在前端统计用户访问来源?

在前端统计用户访问来源是一个常见的需求&#xff0c;通过获取访问来源信息&#xff0c;我们可以了解用户是通过直接访问、搜索引擎、外部链接等途径进入我们的网站或应用。下面是一个详细的介绍&#xff0c;包括方法和实现步骤。 一、获取HTTP Referer HTTP Referer是HTTP请…

MySQL进阶(回望MySQL)——从数据资产谈起 MySQL的DOS命令、常用数据类型、SQL语句的分类 SQL函数

前言 SQL&#xff08;Structured Query Language&#xff09;是一种用于管理关系型数据库的标准化语言&#xff0c;它用于定义、操作和管理数据库中的数据。SQL是一种通用的语言&#xff0c;可以用于多种关系型数据库管理系统&#xff08;RDBMS&#xff09;&#xff0c;如MySQ…

【开发篇】十八、SpringBoot整合ActiveMQ

文章目录 1、安装ActiveMQ2、整合3、发送消息到队列4、使用消息监听器对消息队列监听5、流程性业务消息消费完转入下一个消息队列6、发布订阅模型 1、安装ActiveMQ docker安装 docker pull webcenter/activemqdocker run -d --name activemq -p 61616:61616 -p 8161:8161 webce…

python每日一练(7)

&#x1f308;write in front&#x1f308; &#x1f9f8;大家好&#xff0c;我是Aileen&#x1f9f8;.希望你看完之后&#xff0c;能对你有所帮助&#xff0c;不足请指正&#xff01;共同学习交流. &#x1f194;本文由Aileen_0v0&#x1f9f8; 原创 CSDN首发&#x1f412; 如…

小程序, 多选项

小程序, 多选项 <view class"my-filter-btnwrap"><block wx:for"{{archiveList}}" wx:key"index"><view class"my-filter-btnitem text-ellipsis {{item.checked ? active : }}" data-index"{{index}}" wx…

阿里云r7服务器内存型CPU采用

阿里云服务器ECS内存型r7实例是第七代内存型实例规格族&#xff0c;CPU采用第三代Intel Xeon可扩展处理器&#xff08;Ice Lake&#xff09;&#xff0c;基频2.7 GHz&#xff0c;全核睿频3.5 GHz&#xff0c;计算性能稳定&#xff0c;CPU内存比1:8&#xff0c;2核16G起步&#…

网站列表页加密:三次请求后返回内容多\r

一、抓包第一次请求 url aHR0cDovL2N5eHcuY24vQ29sdW1uLmFzcHg/Y29saWQ9MTA抓包&#xff0c;需要清理浏览器cookie&#xff0c;或者无痕模式打开网址&#xff0c;否则返回的包不全&#xff0c;依照下图中的第一个包进行requests请求 第一次请求后返回 <!DOCTYPE html>…

puppeteer

目录 介绍启动方法功能一、爬虫优势如何实现爬虫小demo 功能二、执行脚本百度搜索脚本demo 功能三、获取cookie&#xff08;这个只能是模拟浏览器当前进入网页的cookie不是平时用的下载的的浏览器的cookie&#xff09;功能四、监控网页&#xff0c;进行性能分析 介绍 puppetee…