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

RocketMQ教程-(5)-功能特性-消费者分类

Apache RocketMQ 支持 PushConsumer 、 SimpleConsumer 以及 PullConsumer 这三种类型的消费者,本文分别从使用方式、实现原理、可靠性重试和适用场景等方面为您介绍这三种类型的消费者。

背景信息​

Apache RocketMQ 面向不同的业务场景提供了不同消费者类型,每种消费者类型的集成方式和控制方式都不一样。了解如下问题,可以帮助您选择更匹配业务场景的消费者类型。

  • 如何实现并发消费:消费者如何使用并发的多线程机制处理消息,以此提高消息处理效率?

  • 如何实现同步、异步消息处理:对于不同的集成场景,消费者获取消息后可能会将消息异步分发到业务逻辑中处理,此时,消息异步化处理如何实现?

  • 如何实现消息可靠处理:消费者处理消息时如何返回响应结果?如何在消息异常情况进行重试,保证消息的可靠处理?

以上问题的具体答案,请参考下文。

功能概述

消息消费流程

如上图所示, Apache RocketMQ 的消费者处理消息时主要经过以下阶段:消息获取--->消息处理--->消费状态提交。

针对以上几个阶段,Apache RocketMQ 提供了不同的消费者类型: PushConsumer 、SimpleConsumer 和 PullConsumer。这几种类型的消费者通过不同的实现方式和接口可满足您在不同业务场景下的消费需求。具体差异如下:

在实际使用场景中,PullConsumer 仅推荐在流处理框架中集成使用,大多数消息收发场景使用 PushConsumer 和 SimpleConsumer 就可以满足需求。

若您的业务场景发生变更,或您当前使用的消费者类型不适合当前业务,您可以选择在 PushConsumer 和SimpleConsumer 之间变更消费者类型。变更消费者类型不影响当前Apache RocketMQ 资源的使用和业务处理。

危险

生产环境中相同的 ConsumerGroup 下严禁混用 PullConsumer 和其他两种消费者,否则会导致消息消费异常。

PushConsumer​

PushConsumers是一种高度封装的消费者类型,消费消息仅通过消费监听器处理业务并返回消费结果。消息的获取、消费状态提交以及消费重试都通过 Apache RocketMQ 的客户端SDK完成。

使用方式

PushConsumer的使用方式比较固定,在消费者初始化时注册一个消费监听器,并在消费监听器内部实现消息处理逻辑。由 Apache RocketMQ 的SDK在后台完成消息获取、触发监听器调用以及进行消息重试处理。

示例代码如下:

// 消费示例:使用PushConsumer消费普通消息。
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "YourTopic";
FilterExpression filterExpression = new FilterExpression("YourFilterTag", FilterExpressionType.TAG);
PushConsumer pushConsumer = provider.newPushConsumerBuilder()// 设置消费者分组。.setConsumerGroup("YourConsumerGroup")// 设置接入点。.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())// 设置预绑定的订阅关系。.setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))// 设置消费监听器。.setMessageListener(new MessageListener() {@Overridepublic ConsumeResult consume(MessageView messageView) {// 消费消息并返回处理结果。return ConsumeResult.SUCCESS;}}).build();

PushConsumer的消费监听器执行结果分为以下三种情况:

  • 返回消费成功:以Java SDK为例,返回ConsumeResult.SUCCESS,表示该消息处理成功,服务端按照消费结果更新消费进度。

  • 返回消费失败:以Java SDK为例,返回ConsumeResult.FAILURE,表示该消息处理失败,需要根据消费重试逻辑判断是否进行重试消费。

  • 出现非预期失败:例如抛异常等行为,该结果按照消费失败处理,需要根据消费重试逻辑判断是否进行重试消费。

PushConsumer 消费消息时,若消息处理逻辑出现预期之外的阻塞导致消息处理一直无法执行成功,SDK会按照消费超时处理强制提交消费失败结果,并按照消费重试逻辑进行处理。消息超时,请参见PushConsumer消费重试策略。

出现消费超时情况时,SDK虽然提交消费失败结果,但是当前消费线程可能仍然无法响应中断,还会继续处理消息。

内部原理

在PushConsumer类型中,消息的实时处理能力是基于SDK内部的典型Reactor线程模型实现的。如下图所示,SDK内置了一个长轮询线程,先将消息异步拉取到SDK内置的缓存队列中,再分别提交到消费线程中,触发监听器执行本地消费逻辑。

PushConsumer原理

 可靠性重试

PushConsumer 消费者类型中,客户端SDK和消费逻辑的唯一边界是消费监听器接口。客户端SDK严格按照监听器的返回结果判断消息是否消费成功,并做可靠性重试。所有消息必须以同步方式进行消费处理,并在监听器接口结束时返回调用结果,不允许再做异步化分发。消息重试具体信息,请参见PushConsumer消费重试策略。

使用PushConsumer消费者消费时,不允许使用以下方式处理消息,否则 Apache RocketMQ 无法保证消息的可靠性。

  • 错误方式一:消息还未处理完成,就提前返回消费成功结果。此时如果消息消费失败,Apache RocketMQ 服务端是无法感知的,因此不会进行消费重试。

  • 错误方式二:在消费监听器内将消息再次分发到自定义的其他线程,消费监听器提前返回消费结果。此时如果消息消费失败,Apache RocketMQ 服务端同样无法感知,因此也不会进行消费重试。

顺序性保障

基于 Apache RocketMQ 顺序消息的定义,如果消费者分组设置了顺序消费模式,则PushConsumer在触发消费监听器时,严格遵循消息的先后顺序。业务处理逻辑无感知即可保证消息的消费顺序。

适用场景

PushConsumer严格限制了消息同步处理及每条消息的处理超时时间,适用于以下场景:

  • 消息处理时间可预估:如果不确定消息处理耗时,经常有预期之外的长时间耗时的消息,PushConsumer的可靠性保证会频繁触发消息重试机制造成大量重复消息。

  • 无异步化、高级定制场景:PushConsumer限制了消费逻辑的线程模型,由客户端SDK内部按最大吞吐量触发消息处理。该模型开发逻辑简单,但是不允许使用异步化和自定义处理流程。

SimpleConsumer

SimpleConsumer 是一种接口原子型的消费者类型,消息的获取、消费状态提交以及消费重试都是通过消费者业务逻辑主动发起调用完成

使用方式

SimpleConsumer 的使用涉及多个接口调用,由业务逻辑按需调用接口获取消息,然后分发给业务线程处理消息,最后按照处理的结果调用提交接口,返回服务端当前消息的处理结果。示例如下:

// 消费示例:使用 SimpleConsumer 消费普通消息,主动获取消息处理并提交。 
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "YourTopic";
FilterExpression filterExpression = new FilterExpression("YourFilterTag", FilterExpressionType.TAG);
SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()// 设置消费者分组。.setConsumerGroup("YourConsumerGroup")// 设置接入点。.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())// 设置预绑定的订阅关系。.setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))// 设置从服务端接受消息的最大等待时间.setAwaitDuration(Duration.ofSeconds(1)).build();
try {// SimpleConsumer 需要主动获取消息,并处理。List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));messageViewList.forEach(messageView -> {System.out.println(messageView);// 消费处理完成后,需要主动调用 ACK 提交消费结果。try {simpleConsumer.ack(messageView);} catch (ClientException e) {logger.error("Failed to ack message, messageId={}", messageView.getMessageId(), e);}});
} catch (ClientException e) {// 如果遇到系统流控等原因造成拉取失败,需要重新发起获取消息请求。logger.error("Failed to receive message", e);
}

SimpleConsumer主要涉及以下几个接口行为:

 可靠性重试

SimpleConsumer消费者类型中,客户端SDK和服务端通过ReceiveMessageAckMessage接口通信。客户端SDK如果处理消息成功则调用AckMessage接口;如果处理失败只需要不回复ACK响应,即可在定义的消费不可见时间到达后触发消费重试流程。更多信息,请参见SimpleConsumer消费重试策略。

顺序性保障

基于 Apache RocketMQ 顺序消息的定义,SimpleConsumer在处理顺序消息时,会按照消息存储的先后顺序获取消息。即需要保持顺序的一组消息中,如果前面的消息未处理完成,则无法获取到后面的消息。

适用场景

SimpleConsumer提供原子接口,用于消息获取和提交消费结果,相对于PushConsumer方式更加灵活。SimpleConsumer适用于以下场景:

  • 消息处理时长不可控:如果消息处理时长无法预估,经常有长时间耗时的消息处理情况。建议使用SimpleConsumer消费类型,可以在消费时自定义消息的预估处理时长,若实际业务中预估的消息处理时长不符合预期,也可以通过接口提前修改。

  • 需要异步化、批量消费等高级定制场景:SimpleConsumer在SDK内部没有复杂的线程封装,完全由业务逻辑自由定制,可以实现异步分发、批量消费等高级定制场景。

  • 需要自定义消费速率:SimpleConsumer是由业务逻辑主动调用接口获取消息,因此可以自由调整获取消息的频率,自定义控制消费速率。

PullConsumer

使用建议​

PushConsumer合理控制消费耗时,避免无限阻塞

对于PushConsumer消费类型,需要严格控制消息的消费耗时,尽量避免出现消息处理超时导致消息重复。如果业务经常会出现一些预期外的长时间耗时的消息,建议使用SimpleConsumer,并设置好消费不可见时间。

相关文章:

RocketMQ教程-(5)-功能特性-消费者分类

Apache RocketMQ 支持 PushConsumer 、 SimpleConsumer 以及 PullConsumer 这三种类型的消费者&#xff0c;本文分别从使用方式、实现原理、可靠性重试和适用场景等方面为您介绍这三种类型的消费者。 背景信息​ Apache RocketMQ 面向不同的业务场景提供了不同消费者类型&…...

Kafka原理剖析

一、简介 Kafka是一个分布式的、分区的、多副本的消息发布-订阅系统&#xff0c;它提供了类似于JMS的特性&#xff0c;但在设计上完全不同&#xff0c;它具有消息持久化、高吞吐、分布式、多客户端支持、实时等特性&#xff0c;适用于离线和在线的消息消费&#xff0c;如常规的…...

word怎么转换成pdf?分享几种转换方法

word怎么转换成pdf&#xff1f;将Word文档转换成PDF文件有几个好处。首先&#xff0c;PDF文件通常比Word文档更容易在不同设备和操作系统上查看和共享。其次&#xff0c;PDF文件通常比Word文档更难以修改&#xff0c;这使得它们在需要保护文件内容的情况下更加安全可靠。最后&a…...

基于XDMA 中断模式的 PCIE3.0 QT上位机与FPGA数据交互架构 提供工程源码和QT上位机源码

目录 1、前言2、我已有的PCIE方案3、PCIE理论4、总体设计思路和方案图像产生、发送、缓存数据处理XDMA简介XDMA中断模式图像读取、输出、显示QT上位机及其源码 5、vivado工程详解6、上板调试验证7、福利&#xff1a;工程代码的获取 1、前言 PCIE&#xff08;PCI Express&#…...

Vue 中通用的 css 列表入场动画效果

css 代码 .gradientAnimation {animation-name: gradient;animation-duration: 0.85s;animation-fill-mode: forwards;opacity: 0; }/* 不带前缀的放到最后 */ keyframes gradient {0% {opacity: 0;transform: translate(-100px, 0px);}100% {opacity: 1;transform: translate…...

微分流形2:流形上的矢量场和张量场

来了来了&#xff0c;切向量&#xff0c;切空间。流形上的所有的线性泛函的集合&#xff0c;注意是函数的集合。然后取流形上的某点p&#xff0c;它的切向量为&#xff0c;线性泛函到实数的映射。没错&#xff0c;是函数到实数的映射&#xff0c;是不是想到了求导。我们要逐渐熟…...

C++数组、向量和列表的练习

运行代码&#xff1a; //C数组、向量和列表的练习 #include"std_lib_facilities.h"int main() try {int ii[10] { 0,1,2,3,4,5,6,7,8,9 };for (int i 0; i < 10; i)//把数组中的每个元素值加2ii[i] 2;vector<int>vv(10);for (int i 0; i < 10; i)vv…...

视频剪辑矩阵分发系统Unable to load FFProbe报错技术处理?

问题一 报错处理 对于视频剪辑矩阵分发系统中出现的“Unable to load FFProbe”报错问题&#xff0c;可以采取以下技术处理措施进行解决。 1.检查系统中是否正确安装了FFProbe工具&#xff0c;并确保其路径正确配置。 2.检查系统环境变量是否正确设置&#xff0c;包括FFPr…...

Docker轻量级可视化工具Portainer

Portainer是一个轻量级的管理UI界面&#xff0c;用于管理Docker容器、镜像、卷和网络。它支持端口映射、容器启动、停止、删除、日志查看等功能&#xff0c;同时也提供了可视化的监控和统计功能&#xff0c;可以快速轻松的管理多个Docker主机。Portainer不需要额外安装依赖&…...

功率放大器在电光调制中的应用有哪些

电光调制是一种利用光电效应将电信号转化为光信号的技术。在实现电光调制的过程中&#xff0c;功率放大器作为一个重要的组件&#xff0c;具有对输入电信号进行放大和控制的功能。本文将介绍功率放大器的基本原理、特点以及在电光调制中的应用。 基本原理 功率放大器是一种能够…...

MyBatis入门程序

1.MyBatis 入门程序开发步骤 SqlSession&#xff1a;代表Java程序和数据库之间的会话。&#xff08;HttpSession是Java程序和浏览器之间的会话&#xff09; SqlSessionFactory&#xff1a;是“生产”SqlSession的“工厂”。 工厂模式&#xff1a;如果创建某一个对象&#xff…...

C++快速切换 头文件和源文件

有没有一种快速的方法 &#xff0c; 将头文件中的声明 直接在源文件中自动写出来&#xff0c; 毕竟头文件中已经有声明了&#xff0c; 我只需要写具体实现就行了&#xff0c;没有必要把声明的部分再敲一遍在 Visual Studio 中&#xff0c;你可以使用快速生成函数定义的功能来实…...

对原型、原型链的理解

在 JavaScript 中是使用构造两数来新建一个对象的&#xff0c;每一个构造函数的内部都有一个 prototype 属性&#xff0c;它的属性值是一个对象&#xff0c;这个对象包含了可以由该构造西数的所有实例共享的属性和方法。当使用构造函数新建一个对象后&#xff0c;在这个对象的内…...

7月26日,每日信息差

1、日本经产省将讨论让消费者负担核电站重启费。若被采用&#xff0c;那么即便是与把源自可再生能源作为卖点的新电力公司签约的消费者&#xff0c;也将负担重启核电站的费用 2、国家发改委&#xff1a;电厂存煤和出力均达历史同期最高水平 3、国家深改委&#xff1a;全国统调…...

git修改已经push后的commit注释

回到倒数第8次提交 git rebase -i HEAD~8修改注释&#xff0c;然后把最前面的pick改成edit 修改注释 git commit --amendrebase确认 git rebase --continue强制提交 git push -f origin master参考&#xff1a;https://blog.csdn.net/qq_16942727/article/details/1260355…...

网络云存储服务器,数据库服务器|PetaExpress

云存储服务器是什么&#xff1f; 云存储服务器是一种在线存储(英语:Cloud storage&#xff09;该模式是将数据存储在通常由第三方托管的多个虚拟服务器上&#xff0c;而不是独家服务器上。 云存储服务器有几种结构 架构方法分为两类&#xff1a;一类是通过服务进行架构&…...

java语法基础--基本数据类型

一、数据类型概括 1、整数类型 2、浮点型 3、布尔类型 4、字符类型 二、数据类型的使用 1、整数类型的使用 超出类型范围 //1.1 定义一个byte类型的变量&#xff0c;并且设置它超过byte类型范围// 如果定义的数值在byte类型范围内&#xff0c;那么就能正常使用&#xff0c;//…...

uniapp 微信小程序 预览pdf方法

效果图&#xff1a; 1、在小程序中 // #ifdef MP */ 是区分运行的环境&#xff0c;在小程序中可使用如下方法uni.downloadFile({url: item.link,//文件地址success: function (res) {var filePath res.tempFilePath;uni.openDocument({filePath: filePath,showMenu: false…...

基于vue+uniapp微信小程序公司企业后勤服务(设备)系统

本系统分为用户和管理员两个角色&#xff0c;其中用户可以注册登陆系统&#xff0c;查看公司公告&#xff0c;查看设备&#xff0c;设备入库&#xff0c;查看通讯录&#xff0c;会议室预约&#xff0c;申请出入&#xff0c;申请请假等功能。管理员可以对员工信息&#xff0c;会…...

Linux命令(54)之blkid

Linux命令之blkid 1.blkid介绍 linux命令blkid被用来查询系统块设备文件系统的类型、卷标、UUID等信息 2.blkid用法 blkid [参数] [设备] blkid参数 参数说明-L <卷标>将卷标转换为设备名-U <UUID>将UUID转换为设备名-p转换设备块-i显示I/O信息 3.实例 3.1.查…...

业务系统对接大模型的基础方案:架构设计与关键步骤

业务系统对接大模型&#xff1a;架构设计与关键步骤 在当今数字化转型的浪潮中&#xff0c;大语言模型&#xff08;LLM&#xff09;已成为企业提升业务效率和创新能力的关键技术之一。将大模型集成到业务系统中&#xff0c;不仅可以优化用户体验&#xff0c;还能为业务决策提供…...

基于Uniapp开发HarmonyOS 5.0旅游应用技术实践

一、技术选型背景 1.跨平台优势 Uniapp采用Vue.js框架&#xff0c;支持"一次开发&#xff0c;多端部署"&#xff0c;可同步生成HarmonyOS、iOS、Android等多平台应用。 2.鸿蒙特性融合 HarmonyOS 5.0的分布式能力与原子化服务&#xff0c;为旅游应用带来&#xf…...

MVC 数据库

MVC 数据库 引言 在软件开发领域,Model-View-Controller(MVC)是一种流行的软件架构模式,它将应用程序分为三个核心组件:模型(Model)、视图(View)和控制器(Controller)。这种模式有助于提高代码的可维护性和可扩展性。本文将深入探讨MVC架构与数据库之间的关系,以…...

Cloudflare 从 Nginx 到 Pingora:性能、效率与安全的全面升级

在互联网的快速发展中&#xff0c;高性能、高效率和高安全性的网络服务成为了各大互联网基础设施提供商的核心追求。Cloudflare 作为全球领先的互联网安全和基础设施公司&#xff0c;近期做出了一个重大技术决策&#xff1a;弃用长期使用的 Nginx&#xff0c;转而采用其内部开发…...

Module Federation 和 Native Federation 的比较

前言 Module Federation 是 Webpack 5 引入的微前端架构方案&#xff0c;允许不同独立构建的应用在运行时动态共享模块。 Native Federation 是 Angular 官方基于 Module Federation 理念实现的专为 Angular 优化的微前端方案。 概念解析 Module Federation (模块联邦) Modul…...

学校时钟系统,标准考场时钟系统,AI亮相2025高考,赛思时钟系统为教育公平筑起“精准防线”

2025年#高考 将在近日拉开帷幕&#xff0c;#AI 监考一度冲上热搜。当AI深度融入高考&#xff0c;#时间同步 不再是辅助功能&#xff0c;而是决定AI监考系统成败的“生命线”。 AI亮相2025高考&#xff0c;40种异常行为0.5秒精准识别 2025年高考即将拉开帷幕&#xff0c;江西、…...

GitFlow 工作模式(详解)

今天再学项目的过程中遇到使用gitflow模式管理代码&#xff0c;因此进行学习并且发布关于gitflow的一些思考 Git与GitFlow模式 我们在写代码的时候通常会进行网上保存&#xff0c;无论是github还是gittee&#xff0c;都是一种基于git去保存代码的形式&#xff0c;这样保存代码…...

保姆级【快数学会Android端“动画“】+ 实现补间动画和逐帧动画!!!

目录 补间动画 1.创建资源文件夹 2.设置文件夹类型 3.创建.xml文件 4.样式设计 5.动画设置 6.动画的实现 内容拓展 7.在原基础上继续添加.xml文件 8.xml代码编写 (1)rotate_anim (2)scale_anim (3)translate_anim 9.MainActivity.java代码汇总 10.效果展示 逐帧…...

VSCode 使用CMake 构建 Qt 5 窗口程序

首先,目录结构如下图: 运行效果: cmake -B build cmake --build build 运行: windeployqt.exe F:\testQt5\build\Debug\app.exe main.cpp #include "mainwindow.h"#include <QAppli...

python基础语法Ⅰ

python基础语法Ⅰ 常量和表达式变量是什么变量的语法1.定义变量使用变量 变量的类型1.整数2.浮点数(小数)3.字符串4.布尔5.其他 动态类型特征注释注释是什么注释的语法1.行注释2.文档字符串 注释的规范 常量和表达式 我们可以把python当作一个计算器&#xff0c;来进行一些算术…...