当前位置: 首页 > 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.查…...

网络编程(Modbus进阶)

思维导图 Modbus RTU&#xff08;先学一点理论&#xff09; 概念 Modbus RTU 是工业自动化领域 最广泛应用的串行通信协议&#xff0c;由 Modicon 公司&#xff08;现施耐德电气&#xff09;于 1979 年推出。它以 高效率、强健性、易实现的特点成为工业控制系统的通信标准。 包…...

手游刚开服就被攻击怎么办?如何防御DDoS?

开服初期是手游最脆弱的阶段&#xff0c;极易成为DDoS攻击的目标。一旦遭遇攻击&#xff0c;可能导致服务器瘫痪、玩家流失&#xff0c;甚至造成巨大经济损失。本文为开发者提供一套简洁有效的应急与防御方案&#xff0c;帮助快速应对并构建长期防护体系。 一、遭遇攻击的紧急应…...

ubuntu搭建nfs服务centos挂载访问

在Ubuntu上设置NFS服务器 在Ubuntu上&#xff0c;你可以使用apt包管理器来安装NFS服务器。打开终端并运行&#xff1a; sudo apt update sudo apt install nfs-kernel-server创建共享目录 创建一个目录用于共享&#xff0c;例如/shared&#xff1a; sudo mkdir /shared sud…...

SCAU期末笔记 - 数据分析与数据挖掘题库解析

这门怎么题库答案不全啊日 来简单学一下子来 一、选择题&#xff08;可多选&#xff09; 将原始数据进行集成、变换、维度规约、数值规约是在以下哪个步骤的任务?(C) A. 频繁模式挖掘 B.分类和预测 C.数据预处理 D.数据流挖掘 A. 频繁模式挖掘&#xff1a;专注于发现数据中…...

DAY 47

三、通道注意力 3.1 通道注意力的定义 # 新增&#xff1a;通道注意力模块&#xff08;SE模块&#xff09; class ChannelAttention(nn.Module):"""通道注意力模块(Squeeze-and-Excitation)"""def __init__(self, in_channels, reduction_rat…...

YSYX学习记录(八)

C语言&#xff0c;练习0&#xff1a; 先创建一个文件夹&#xff0c;我用的是物理机&#xff1a; 安装build-essential 练习1&#xff1a; 我注释掉了 #include <stdio.h> 出现下面错误 在你的文本编辑器中打开ex1文件&#xff0c;随机修改或删除一部分&#xff0c;之后…...

多模态商品数据接口:融合图像、语音与文字的下一代商品详情体验

一、多模态商品数据接口的技术架构 &#xff08;一&#xff09;多模态数据融合引擎 跨模态语义对齐 通过Transformer架构实现图像、语音、文字的语义关联。例如&#xff0c;当用户上传一张“蓝色连衣裙”的图片时&#xff0c;接口可自动提取图像中的颜色&#xff08;RGB值&…...

2021-03-15 iview一些问题

1.iview 在使用tree组件时&#xff0c;发现没有set类的方法&#xff0c;只有get&#xff0c;那么要改变tree值&#xff0c;只能遍历treeData&#xff0c;递归修改treeData的checked&#xff0c;发现无法更改&#xff0c;原因在于check模式下&#xff0c;子元素的勾选状态跟父节…...

拉力测试cuda pytorch 把 4070显卡拉满

import torch import timedef stress_test_gpu(matrix_size16384, duration300):"""对GPU进行压力测试&#xff0c;通过持续的矩阵乘法来最大化GPU利用率参数:matrix_size: 矩阵维度大小&#xff0c;增大可提高计算复杂度duration: 测试持续时间&#xff08;秒&…...

IoT/HCIP实验-3/LiteOS操作系统内核实验(任务、内存、信号量、CMSIS..)

文章目录 概述HelloWorld 工程C/C配置编译器主配置Makefile脚本烧录器主配置运行结果程序调用栈 任务管理实验实验结果osal 系统适配层osal_task_create 其他实验实验源码内存管理实验互斥锁实验信号量实验 CMISIS接口实验还是得JlINKCMSIS 简介LiteOS->CMSIS任务间消息交互…...