当前位置: 首页 > news >正文

四、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,并清除消息延迟属性,从新进行消息投递

相关文章:

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

Producer发送普通消息的方式 1.同步发送消息 同步消息代表发送端发送消息到broker之后&#xff0c;等待消息发送结果后&#xff0c;再次发送消息 实现步骤 创建生产端&#xff0c;声明在哪个生产组注册NameServer地址构建Message实体&#xff0c;指定topic、tag、body启动…...

idea自定义 postfix completion提高编码效率

postfix completion的使用 详情见&#xff1a; https://www.cnblogs.com/expiator/p/17380495.html 自定义 postfix completion List、 String 初始化list&#xff1a; key: list表达式&#xff1a; List<$EXPR$> $END$List new ArrayList<>();字符串判空&…...

解锁学习电路设计的正确姿势!

...

【Linux】 ps命令使用

作为一个后端的程序员&#xff0c;我们经常用到ps -ef | grep XXX 到底什么事ps呢。 下面我们一起学习一下吧、 ps &#xff08;英文全拼&#xff1a;process status&#xff09;命令用于显示当前进程的状态&#xff0c;类似于 windows 的任务管理器。 ps命令 -Linux手册页 …...

打造高效的分布式爬虫系统:利用Scrapy框架实现

在大数据时代的今天&#xff0c;爬虫系统成为了获取和分析海量数据的重要工具。本文将介绍如何使用Scrapy框架来构建一个高效的分布式爬虫系统&#xff0c;以加速数据采集过程和提高系统的可扩展性。 Scrapy框架简介 Scrapy是一个基于Python的强大的开源网络爬虫框架&#xff…...

SpringCloud组件Ribbon的IRule的问题排查

最近很久没有写文章啦&#xff0c;刚好遇到了一个问题&#xff0c;其实问题也挺简单&#xff0c;但是还是得对源码有一定了解才能够发现。 最近在实现一个根据请求流量的标签&#xff0c;将请求转发到对应的节点&#xff0c;其实和俗称的灰度请求有点相似&#xff0c; 实现思…...

比较完整一些chatGPT项目代码(权威)

https://gitee.com/zccbbg/chatgpt-springboot-service yml中的配置文件无法读取&#xff0c;前端访问比较困难。...

Python - 生成二维码、条形码

二维码 import qrcode# 要生成的文本或链接 data "要生成的文本或链接"# 创建QR码对象 qr qrcode.QRCode(version1, # 版本号&#xff0c;通常设置为1error_correctionqrcode.constants.ERROR_CORRECT_L, # 错误修正级别box_size10, # 每个小方块的像素大小bor…...

8+纯生信,多组机器学习+分型探讨黑色素瘤发文思路。

今天给同学们分享一篇泛癌多组机器学习分型的生信文章“Comprehensive characterisation of immunogenic cell death in melanoma revealing the association with prognosis and tumor immune microenvironment”&#xff0c;这篇文章于2022年9月23日发表在Front Immunol 期刊…...

GPU高性能面试-写一个ReduceKernel

要求写一个reduceKernel 要求给出Kerne的完整调用: 1. 进行一维reduce 可以写一个最基础的&#xff0c;仅仅实现基础功能就行 使用share mem进行功能优化 使用shuffles指令完成block reduce操作 2.实现二维reduce...

深入探索STARK的安全性和可靠性——STARKs全面安全分析

1. 引言 non-interactive STARKs&#xff0c;起源于Interactive Oracle Proofs (IOPs)&#xff0c;然后通过random oracle模式转换为非交互式。StarkWare团队 ethSTARK Documentation – Version 1.2&#xff08;2023年7月&#xff09;论文做了更新&#xff0c;给出了完整具体…...

WPF 控件分辨率自适应问题

WPF 控件分辨率自适应时&#xff0c;我首先想到的是使用ViewBox控件来做分辨率自适应。 ViewBox这个控件通常和其他控件结合起来使用&#xff0c;是WPF中非常有用的控件。定义一个内容容器。ViewBox组件的作用是拉伸或延展位于其中的组件&#xff0c;以填满可用空间&#xff0…...

CANoe创建仿真工程

CANoe创建仿真工程 写在前面仿真工程的创建创建工程添加CAN数据库添加系统变量创建面板创建网络节点为节点添加代码工程运行测试总结 写在前面 Canoe的安装不是特别方便&#xff0c;我是参加了松勤的培训课程&#xff0c;不仅需要安装软件还需要安装驱动&#xff0c;刚刚学习的…...

Scanner 输入回车跳不出循环的解决方法

题目要求&#xff1a; 输入一行内容包含字符串和数字&#xff0c;将字符串与数字分别提取。 解决方法&#xff1a; 可以使用两个Scanner对象&#xff0c;一个用来键入数据&#xff0c;另外一个用来对数据进行操作&#xff0c;以此来解决输入“回车”跳不出while循环的问题。 i…...

docker 启动 mysql 通过防火墙设置端口无法访问解决方案

1、问题描述&#xff1a;通过 docker compose 启动mysql服务&#xff0c;然而在防火墙添加了3306端口后却无法访问&#xff0c;但是关闭防火墙后又可以访问mysql数据库。 解决方案&#xff1a; 重启 docker 后解决&#xff1a;systemctl restart docker 如果没有解决问题则执…...

智能制造优化,RFID生产线管理系统解决方案

一、背景介绍 随着全球经济的发展&#xff0c;传统制造业面临着越来越高的成本和低利润的挑战&#xff0c;为了提升企业的整体利润率&#xff0c;优化管理流程成为必要的手段之一&#xff0c;在传统的制造企业中&#xff0c;生产线通常采用单件流生产模式&#xff0c;但这种模…...

【Mybatis】基于Mybatis插件+注解,实现敏感数据自动加解密

一、介绍 业务场景中经常会遇到诸如用户手机号&#xff0c;身份证号&#xff0c;银行卡号&#xff0c;邮箱&#xff0c;地址&#xff0c;密码等等信息&#xff0c;属于敏感信息&#xff0c;需要保存在数据库中。而很多公司会会要求对数据库中的此类数据进行加密存储。 敏感数据…...

【特纳斯电子】基于物联网的指纹密码锁系统设计-实物设计

资料下载链接&#xff1a;基于物联网的指纹密码锁系统设计-实物设计 - 电子校园网 编号&#xff1a; T3732205M-SW 设计简介&#xff1a; 本设计是基于单片机的指纹密码锁&#xff0c;主要实现以下功能&#xff1a; 1、可通过密码解锁 2、可通过云平台解锁 3、可通过指纹解…...

【牛客面试必刷TOP101】Day9.BM37 二叉搜索树的最近公共祖先和BM42 用两个栈实现队列

作者简介&#xff1a;大家好&#xff0c;我是未央&#xff1b; 博客首页&#xff1a;未央.303 系列专栏&#xff1a;牛客面试必刷TOP101 每日一句&#xff1a;人的一生&#xff0c;可以有所作为的时机只有一次&#xff0c;那就是现在&#xff01;&#xff01;&#xff01;&…...

10.12 校招 实习 内推 面经

绿*泡*泡&#xff1a; neituijunsir 交流裙 &#xff0c;内推/实习/校招汇总表格 1、校招 | 2024届秋招&#xff0c;美团哪些校招岗位最缺人&#xff1f;&#xff08;内推&#xff09; 校招 | 2024届秋招&#xff0c;美团哪些校招岗位最缺人&#xff1f;&#xff08;内推&…...

树莓派超全系列教程文档--(61)树莓派摄像头高级使用方法

树莓派摄像头高级使用方法 配置通过调谐文件来调整相机行为 使用多个摄像头安装 libcam 和 rpicam-apps依赖关系开发包 文章来源&#xff1a; http://raspberry.dns8844.cn/documentation 原文网址 配置 大多数用例自动工作&#xff0c;无需更改相机配置。但是&#xff0c;一…...

云启出海,智联未来|阿里云网络「企业出海」系列客户沙龙上海站圆满落地

借阿里云中企出海大会的东风&#xff0c;以**「云启出海&#xff0c;智联未来&#xff5c;打造安全可靠的出海云网络引擎」为主题的阿里云企业出海客户沙龙云网络&安全专场于5.28日下午在上海顺利举办&#xff0c;现场吸引了来自携程、小红书、米哈游、哔哩哔哩、波克城市、…...

mongodb源码分析session执行handleRequest命令find过程

mongo/transport/service_state_machine.cpp已经分析startSession创建ASIOSession过程&#xff0c;并且验证connection是否超过限制ASIOSession和connection是循环接受客户端命令&#xff0c;把数据流转换成Message&#xff0c;状态转变流程是&#xff1a;State::Created 》 St…...

Python爬虫实战:研究feedparser库相关技术

1. 引言 1.1 研究背景与意义 在当今信息爆炸的时代,互联网上存在着海量的信息资源。RSS(Really Simple Syndication)作为一种标准化的信息聚合技术,被广泛用于网站内容的发布和订阅。通过 RSS,用户可以方便地获取网站更新的内容,而无需频繁访问各个网站。 然而,互联网…...

跨链模式:多链互操作架构与性能扩展方案

跨链模式&#xff1a;多链互操作架构与性能扩展方案 ——构建下一代区块链互联网的技术基石 一、跨链架构的核心范式演进 1. 分层协议栈&#xff1a;模块化解耦设计 现代跨链系统采用分层协议栈实现灵活扩展&#xff08;H2Cross架构&#xff09;&#xff1a; 适配层&#xf…...

sqlserver 根据指定字符 解析拼接字符串

DECLARE LotNo NVARCHAR(50)A,B,C DECLARE xml XML ( SELECT <x> REPLACE(LotNo, ,, </x><x>) </x> ) DECLARE ErrorCode NVARCHAR(50) -- 提取 XML 中的值 SELECT value x.value(., VARCHAR(MAX))…...

TRS收益互换:跨境资本流动的金融创新工具与系统化解决方案

一、TRS收益互换的本质与业务逻辑 &#xff08;一&#xff09;概念解析 TRS&#xff08;Total Return Swap&#xff09;收益互换是一种金融衍生工具&#xff0c;指交易双方约定在未来一定期限内&#xff0c;基于特定资产或指数的表现进行现金流交换的协议。其核心特征包括&am…...

【Zephyr 系列 10】实战项目:打造一个蓝牙传感器终端 + 网关系统(完整架构与全栈实现)

🧠关键词:Zephyr、BLE、终端、网关、广播、连接、传感器、数据采集、低功耗、系统集成 📌目标读者:希望基于 Zephyr 构建 BLE 系统架构、实现终端与网关协作、具备产品交付能力的开发者 📊篇幅字数:约 5200 字 ✨ 项目总览 在物联网实际项目中,**“终端 + 网关”**是…...

HashMap中的put方法执行流程(流程图)

1 put操作整体流程 HashMap 的 put 操作是其最核心的功能之一。在 JDK 1.8 及以后版本中&#xff0c;其主要逻辑封装在 putVal 这个内部方法中。整个过程大致如下&#xff1a; 初始判断与哈希计算&#xff1a; 首先&#xff0c;putVal 方法会检查当前的 table&#xff08;也就…...

基于SpringBoot在线拍卖系统的设计和实现

摘 要 随着社会的发展&#xff0c;社会的各行各业都在利用信息化时代的优势。计算机的优势和普及使得各种信息系统的开发成为必需。 在线拍卖系统&#xff0c;主要的模块包括管理员&#xff1b;首页、个人中心、用户管理、商品类型管理、拍卖商品管理、历史竞拍管理、竞拍订单…...