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

rabbitmq的消息应答

消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成
了部分突然它挂掉了,会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息,便立即将该消
息标记为删除。在这种情况下,突然有个消费者挂掉了,我们将丢失正在处理的消息。以及后续
发送给该消费这的消息,因为它无法接收到。
为了保证消息在发送过程中不丢失,rabbitmq 引入消息应答机制,消息应答就是: 消费者在接
收到消息并且处理该消息之后,告诉 rabbitmq 它已经处理了,rabbitmq 可以把该消息删除了。

消息应答的方法 

Channel.basicAck(用于肯定确认)

RabbitMQ 已知道该消息并且成功的处理消息,可以将其丢弃了

Channel.basicNack(用于否定确认)

Channel.basicReject(用于否定确认)

与 Channel.basicNack 相比少一个参数
不处理该消息了直接拒绝,可以将其丢弃了

自动应答 

消息发送后立即被认为已经传送成功,这种模式需要在 高吞吐量和数据传输安全性方面做权
,因为这种模式如果消息在接收到之前,消费者那边出现连接或者 channel 关闭,那么消息就丢
失了,当然另一方面这种模式消费者那边可以传递过载的消息, 没有对传递的消息数量进行限制
当然这样有可能使得消费者这边由于接收太多还来不及处理的消息,导致这些消息的积压,最终
使得内存耗尽,最终这些消费者线程被操作系统杀死, 所以这种模式仅适用在消费者可以高效并
以某种速率能够处理这些消息的情况下使用。默认消息采用的是自动应答.

手动应答

手动应答的好处是可以批量应答并且减少网络拥堵multiple 的 true 和 false 代表不同意思

true 代表批量应答 channel 上未应答的消息
比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8 那么此时
5-8 的这些还未应答的消息都会被确认收到消息应答
false 同上面相比
只会应答 tag=8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答

消费者1 

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;public class Worker01 {private static final String QUEUE_NAME="hello";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();System.out.println("C1 等待接收消息处理时间较短");DeliverCallback deliverCallback=(consumerTag, delivery)->{String receivedMessage = new String(delivery.getBody());System.out.println("接收到消息:"+receivedMessage);try {// 睡眠1秒Thread.sleep(1000*1);} catch (InterruptedException e) {throw new RuntimeException(e);}// 消息标记tag,2.false代表只应答接收到的那个传递的消息,true为应答所有消息包括传递过来的消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback=(consumerTag)->{System.out.println(consumerTag+"消费者取消消费接口回调逻辑");};System.out.println("C1 消费者启动等待消费......");boolean autoAck=false;channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,cancelCallback);}
}

 消费者2


import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;public class Worker02 {private static final String QUEUE_NAME="hello";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();System.out.println("C2 等待接收消息处理时间较长");DeliverCallback deliverCallback=(consumerTag, delivery)->{String receivedMessage = new String(delivery.getBody());System.out.println("接收到消息:"+receivedMessage);try {// 睡眠30秒Thread.sleep(1000*30);} catch (InterruptedException e) {throw new RuntimeException(e);}// 消息标记tag,2.false代表只应答接收到的那个传递的消息,true为应答所有消息包括传递过来的消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback=(consumerTag)->{System.out.println(consumerTag+"消费者取消消费接口回调逻辑");};System.out.println("C2 消费者启动等待消费......");boolean autoAck=false;channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,cancelCallback);}
}

生产者 

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {private final static String QUEUE_NAME = "hello";public static void main(String[] args) throws Exception {//创建一个连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("43.139.59.28");factory.setUsername("guest");factory.setPassword("guest");//channel 实现了自动 close 接口 自动关闭 不需要显示关闭try(Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {/*** 生成一个队列* 1.队列名称* 2.队列里面的消息是否持久化 默认消息存储在内存中* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除* 5.其他参数*/channel.queueDeclare(QUEUE_NAME,false,false,false,null);String message="hello world!!!!";/*** 发送一个消息* 1.发送到那个交换机* 2.路由的 key 是哪个* 3.其他的参数信息* 4.发送消息的消息体*/for (int i = 0; i < 10; i++) {channel.basicPublish("",QUEUE_NAME,null,message.getBytes());}System.out.println("消息发送完毕");}}
}

结果

在发送者发送消息 ,发出消息之后的把 C2 消费者停掉,按理说该 C2 来处理该消息,但是
由于它处理时间较长,在还未处理完,也就是说 C2 还没有执行 ack 代码的时候,C2 被停掉了,
此时会看到消息被 C1 接收到了,说明消息  被重新入队,然后分配给能处理消息的 C1 处理了 

消息自动重新入队

如果消费者由于某些原因失去连接(其通道已关闭,连接已关闭或 TCP 连接丢失),导致消息
未发送 ACK 确认,RabbitMQ 将了解到消息未完全处理,并将对其重新排队。如果此时其他消费者
可以处理,它将很快将其重新分发给另一个消费者。这样,即使某个消费者偶尔死亡,也可以确
保不会丢失任何消息。

 

相关文章:

rabbitmq的消息应答

消费者完成一个任务可能需要一段时间&#xff0c;如果其中一个消费者处理一个长的任务并仅只完成 了部分突然它挂掉了&#xff0c;会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息&#xff0c;便立即将该消 息标记为删除。在这种情况下&#xff0c;突然有个消费者挂掉了…...

如何重置树莓派 Pico(重置外围设备失败)

有时候需要重置树莓派 Pico&#xff0c;一种方法是按住 Pico 上的“BOOTSEL”按钮再插入 USB&#xff1b;或者用按钮连接“RUN”和“GND”针脚&#xff0c;然后同时按下这个按钮和“BOOTSEL”按钮。这样就可以进入 USB 模式&#xff0c;这样从一定程度进行了重置。 但是这种方…...

LaWGPT基于中文法律知识的大语言模型_初步安装

准备代码&#xff0c;创建环境 # 下载代码 git clone gitgithub.com:pengxiao-song/LaWGPT.git cd LaWGPT# 创建环境 conda create -n lawgpt python3.10 -y conda activate lawgpt国内网络环境问题。你可以把requirements.txt里面的github.com替换成kgithub.com&#xff08;这…...

一文学会sklearn中的交叉验证方法,cross_validate和KFlod实战案例

前言 在机器学习中&#xff0c;我们经常需要评估模型的性能。而为了准确评估模型的性能&#xff0c;我们需要使用一种有效的评估方法。五折交叉验证&#xff08;5-fold cross-validation&#xff09;就是其中一种常用的模型评估方法&#xff0c;用于评估机器学习模型的性能和泛…...

《面试1v1》ElasticSearch倒排索引

&#x1f345; 作者简介&#xff1a;王哥&#xff0c;CSDN2022博客总榜Top100&#x1f3c6;、博客专家&#x1f4aa; &#x1f345; 技术交流&#xff1a;定期更新Java硬核干货&#xff0c;不定期送书活动 &#x1f345; 王哥多年工作总结&#xff1a;Java学习路线总结&#xf…...

基于架构的软件开发方法

基于架构的软件开发方法 基于架构的软件开发方法是由架构驱动的&#xff0c;即指由构成体系结构的商业、质量和功能需求的组合驱动的。使用ABSD 方法&#xff0c;设计活动可以从项目总体功能框架明确就开始&#xff0c;这意味着需求抽取和分析还没有完成(甚至远远没有完成)&am…...

实战篇之基于二进制思想的用户标签系统(Mysql+SpringBoot)

一&#xff1a; 计算机中的二进制 计算机以二进制表示数据&#xff0c;以表示电路中的正反。在二进制下&#xff0c;一个位只有 0 和 1 。逢二进一 位。类似十进制下&#xff0c;一个位只有 0~9 。逢十进一位。 二&#xff1a; 进制常用运算 &#xff08;位运算&#xff09;…...

Ansible 进阶

Ansible 进阶 ⤴️Ansible 入门看这篇文章⤵️Ansible 实战看这篇文章 一.Ansible 中的 Playbook 1.1 Playbook 介绍 如下图&#xff0c;ansible 在整个管理过程中使用 playbook 的大体流程。 Playbook 中包含多个 role&#xff0c;每个 role 对应于在远程主机完成某个比较复…...

滴滴Ceph分布式存储系统优化之锁优化

摘自&#xff1a;https://mp.weixin.qq.com/s/oWujGOLLGItu1Bv5AuO0-A 2020-09-02 21:45 0.引言 Ceph是国际知名的开源分布式存储系统&#xff0c;在工业界和学术界都有着重要的影响。Ceph的架构和算法设计发表在国际系统领域顶级会议OSDI、SOSP、SC等上。Ceph社区得到Red Hat…...

flutter开发实战-MethodChannel实现flutter与iOS双向通信

flutter开发实战-MethodChannel实现flutter与iOS双向通信 最近开发中需要iOS与flutter实现通信&#xff0c;这里使用的MethodChannel 如果需要flutter与Android实现双向通信&#xff0c;请看 https://blog.csdn.net/gloryFlow/article/details/132218837 这部分与https://bl…...

华为、阿里巴巴、字节跳动 100+ Python 面试问题总结(七)

系列文章目录 个人简介&#xff1a;机电专业在读研究生&#xff0c;CSDN内容合伙人&#xff0c;博主个人首页 Python面试专栏&#xff1a;《Python面试》此专栏面向准备面试的2024届毕业生。欢迎阅读&#xff0c;一起进步&#xff01;&#x1f31f;&#x1f31f;&#x1f31f; …...

K8S系列一:概念入门

写在前面 本文组织方式&#xff1a; K8S的架构、作用和目的。需要首先对K8S整体有所了解。 K8S是什么&#xff1f; 为什么是K8S&#xff1f; K8S怎么做&#xff1f; K8S的重要概念&#xff0c;即K8S的API对象。要学习和使用K8S必须知道和掌握的几个对象。 Pod 实例 Volume 数…...

QT- QLineEdite设置自动补全功能,并修改自动补全的样式

#include <QApplication> #include <QLineEdit> #include <QCompleter> #include <QStringListModel> #include <QMainWindow>int main(int argc, char *argv[]) {QApplication app(argc, argv);// 创建一个字符串列表模型QStringListModel mode…...

解决Adobe Flash Player已被屏蔽

问题&#xff1a;该插件不支持 原因&#xff1a;现在浏览器默认禁用flash 博主当前使用的是谷歌浏览器Chrome 2个主要方法都已经失效 搜索一圈后&#xff0c;之前博客给出的2个主要方法都已经失效。 1、flash.cn 下载本地播放器 2、在chrome中打开flash的禁用开关 2023年解…...

【Spring专题】Spring之Bean的生命周期源码解析——阶段二(IOC之实例化)

目录 前言阅读准备阅读指引阅读建议 课程内容一、SpringIOC之实例化1.1 简单回顾1.2 概念回顾1.3 核心方法讲解 二、方法讲解2.1 AbstractBeanFactory#getMergedLocalBeanDefinition&#xff1a;合并BeanDefinition2.2 AbstractAutowireCapableBeanFactory#createBean&#xff…...

YOLOv8目标检测算法

YOLOv8目标检测算法相较于前几代YOLO系列算法具有如下的几点优势&#xff1a; 更友好的安装/运行方式速度更快、准确率更高新的backbone&#xff0c;将YOLOv5中的C3更换为C2FYOLO系列第一次尝试使用anchor-free新的损失函数 YOLOv8简介 YOLOv8 是 Ultralytics 公司继 YOLOv5…...

uniapp条件编译

uniapp条件编译 uni-app 将已经将可以跨平台兼容处理的组件及 API 等进行了封装&#xff0c;但是部分平台的特性无法跨平台。 由此&#xff0c;uni-app 提供了条件编译的方案&#xff0c;来处理不同平台的特定或差异。 写法 #ifdef&#xff1a; 仅在某平台存在#ifndef&#xf…...

2023年国赛数学建模思路 - 复盘:光照强度计算的优化模型

文章目录 0 赛题思路1 问题要求2 假设约定3 符号约定4 建立模型5 模型求解6 实现代码 建模资料 0 赛题思路 &#xff08;赛题出来以后第一时间在CSDN分享&#xff09; https://blog.csdn.net/dc_sinor?typeblog 1 问题要求 现在已知一个教室长为15米&#xff0c;宽为12米&…...

volte端到端问题分析(一)

1、MME专载保持功能验证 **描述&#xff1a;**当无线环境较差时&#xff0c;有可能由于“Radio_Connection_with_UE_Lost” 原因造成的VoLTE通话掉话&#xff0c;如果UE发生RRC重建成功&#xff0c;手机将不会掉话。 对MME1202进行功能验证&#xff1a;开启后&#xff0c;MME专…...

微信小程序(原生)搜索功能实现

一、效果图 二、代码 wxml <van-searchvalue"{{ keyword }}"shape"round"background"#000"placeholder"请输入关键词"use-action-slotbind:change"onChange"bind:search"onSearch"bind:clear"onClear&q…...

装饰模式(Decorator Pattern)重构java邮件发奖系统实战

前言 现在我们有个如下的需求&#xff0c;设计一个邮件发奖的小系统&#xff0c; 需求 1.数据验证 → 2. 敏感信息加密 → 3. 日志记录 → 4. 实际发送邮件 装饰器模式&#xff08;Decorator Pattern&#xff09;允许向一个现有的对象添加新的功能&#xff0c;同时又不改变其…...

K8S认证|CKS题库+答案| 11. AppArmor

目录 11. AppArmor 免费获取并激活 CKA_v1.31_模拟系统 题目 开始操作&#xff1a; 1&#xff09;、切换集群 2&#xff09;、切换节点 3&#xff09;、切换到 apparmor 的目录 4&#xff09;、执行 apparmor 策略模块 5&#xff09;、修改 pod 文件 6&#xff09;、…...

MFC内存泄露

1、泄露代码示例 void X::SetApplicationBtn() {CMFCRibbonApplicationButton* pBtn GetApplicationButton();// 获取 Ribbon Bar 指针// 创建自定义按钮CCustomRibbonAppButton* pCustomButton new CCustomRibbonAppButton();pCustomButton->SetImage(IDB_BITMAP_Jdp26)…...

Linux-07 ubuntu 的 chrome 启动不了

文章目录 问题原因解决步骤一、卸载旧版chrome二、重新安装chorme三、启动不了&#xff0c;报错如下四、启动不了&#xff0c;解决如下 总结 问题原因 在应用中可以看到chrome&#xff0c;但是打不开(说明&#xff1a;原来的ubuntu系统出问题了&#xff0c;这个是备用的硬盘&a…...

laravel8+vue3.0+element-plus搭建方法

创建 laravel8 项目 composer create-project --prefer-dist laravel/laravel laravel8 8.* 安装 laravel/ui composer require laravel/ui 修改 package.json 文件 "devDependencies": {"vue/compiler-sfc": "^3.0.7","axios": …...

Python ROS2【机器人中间件框架】 简介

销量过万TEEIS德国护膝夏天用薄款 优惠券冠生园 百花蜂蜜428g 挤压瓶纯蜂蜜巨奇严选 鞋子除臭剂360ml 多芬身体磨砂膏280g健70%-75%酒精消毒棉片湿巾1418cm 80片/袋3袋大包清洁食品用消毒 优惠券AIMORNY52朵红玫瑰永生香皂花同城配送非鲜花七夕情人节生日礼物送女友 热卖妙洁棉…...

【7色560页】职场可视化逻辑图高级数据分析PPT模版

7种色调职场工作汇报PPT&#xff0c;橙蓝、黑红、红蓝、蓝橙灰、浅蓝、浅绿、深蓝七种色调模版 【7色560页】职场可视化逻辑图高级数据分析PPT模版&#xff1a;职场可视化逻辑图分析PPT模版https://pan.quark.cn/s/78aeabbd92d1...

短视频矩阵系统文案创作功能开发实践,定制化开发

在短视频行业迅猛发展的当下&#xff0c;企业和个人创作者为了扩大影响力、提升传播效果&#xff0c;纷纷采用短视频矩阵运营策略&#xff0c;同时管理多个平台、多个账号的内容发布。然而&#xff0c;频繁的文案创作需求让运营者疲于应对&#xff0c;如何高效产出高质量文案成…...

苹果AI眼镜:从“工具”到“社交姿态”的范式革命——重新定义AI交互入口的未来机会

在2025年的AI硬件浪潮中,苹果AI眼镜(Apple Glasses)正在引发一场关于“人机交互形态”的深度思考。它并非简单地替代AirPods或Apple Watch,而是开辟了一个全新的、日常可接受的AI入口。其核心价值不在于功能的堆叠,而在于如何通过形态设计打破社交壁垒,成为用户“全天佩戴…...

算术操作符与类型转换:从基础到精通

目录 前言&#xff1a;从基础到实践——探索运算符与类型转换的奥秘 算术操作符超级详解 算术操作符&#xff1a;、-、*、/、% 赋值操作符&#xff1a;和复合赋值 单⽬操作符&#xff1a;、--、、- 前言&#xff1a;从基础到实践——探索运算符与类型转换的奥秘 在先前的文…...