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

rabbitMq举例

news 来源：原创 2025/6/21 1:28:27

新来个技术总监，把 RabbitMQ 讲的那叫一个透彻，佩服！

生产者

代码举例

public String sendMsg(final String exchangeName,final String routingKey,final String msg) {}

/*** 发送消息* @param exchangeName	exchangeName* @param routingKey	routingKey* @param msg	mq message* @return msgId*/public String sendMsg(final String exchangeName,final String routingKey,final String msg) {final CorrelationData correlationDataInfo = new CorrelationData();final String msgId = String.join(":", "yl:dps", exchangeName, routingKey,String.valueOf(IdUtil.getSnowflake(1, 3).nextId()));correlationDataInfo.setId(msgId);//send over callback lograbbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {if(correlationData!=null) {final String id = correlationData.getId();if (ack) {log.info("消息投递成功,消息Id[{}]【{}】", id, JSON.toJSONString(correlationData));} else {log.error("消息投递失败,消息Id[{}] [{}]", id, cause);}}else {log.error("消息投递失败，correlationData为null!");}});rabbitTemplate.convertAndSend(exchangeName, routingKey, msg, correlationDataInfo);

分析代码

这段代码的功能是向 RabbitMQ 发送消息，并且在发送消息后通过回调函数来确认消息的投递状态。下面是对代码的详细分析：

1. 方法签名

java

public String sendMsg(final String exchangeName, final String routingKey, final String msg)

该方法接收三个参数：
- exchangeName：消息发送的交换机名称。
- routingKey：消息的路由键，用来确定消息将被路由到哪个队列。
- msg：待发送的消息内容。

2. CorrelationData 创建与消息 ID 设置

java

final CorrelationData correlationDataInfo = new CorrelationData(); final String msgId = String.join(":", "yl:dps", exchangeName, routingKey, String.valueOf(IdUtil.getSnowflake(1, 3).nextId())); correlationDataInfo.setId(msgId);

CorrelationData：RabbitMQ 提供的一个类，用于设置和传递消息的相关信息，主要用来在消息确认回调中标识消息。
生成 msgId：msgId 是通过拼接一个前缀 "yl:dps"、交换机名 exchangeName、路由键 routingKey，以及通过 IdUtil.getSnowflake(1, 3).nextId() 生成的雪花算法（Snowflake）ID来创建的唯一标识符。这个 ID 用来唯一标识这条消息。
setId(msgId)：将生成的 msgId 设置到 correlationDataInfo 对象中。

3. 设置回调函数

java

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { if(correlationData != null) { final String id = correlationData.getId(); if (ack) { log.info("消息投递成功,消息Id[{}]【{}】", id, JSON.toJSONString(correlationData)); } else { log.error("消息投递失败,消息Id[{}] [{}]", id, cause); } } else { log.error("消息投递失败，correlationData为null!"); } });

setConfirmCallback：通过 rabbitTemplate.setConfirmCallback() 方法设置消息确认回调函数。在消息发送之后，RabbitMQ 会调用该回调函数来通知消息是否成功投递到交换机。
参数说明：
- correlationData：携带消息相关信息的对象（这里是 correlationDataInfo），包含了消息的 ID 等。
- ack：布尔值，表示消息是否成功投递（true 表示成功，false 表示失败）。
- cause：如果 ack 为 false，cause 会包含失败的原因。
回调内容：
- 如果 correlationData 不为空：
  - 如果 ack 为 true，则表示消息投递成功，记录日志。
  - 如果 ack 为 false，则表示消息投递失败，记录失败日志，并包含失败原因。
- 如果 correlationData 为 null，记录错误日志。

4. 发送消息

java

rabbitTemplate.convertAndSend(exchangeName, routingKey, msg, correlationDataInfo);

convertAndSend：该方法用于将消息发送到指定的交换机（exchangeName）、路由键（routingKey）和消息内容（msg）。同时，将 correlationDataInfo（包含 msgId）附带发送，以便在回调中进行确认和记录。
rabbitTemplate 是 RabbitTemplate 对象，用来发送和接收消息。它是 Spring AMQP 的核心组件，封装了与 RabbitMQ 的交互。

5. 返回消息 ID

return msgId;

返回生成的消息 ID。这个 ID 可以在外部被用于追踪消息投递的状态。

总结：

这段代码的主要作用是：

生成一个唯一的消息 ID，包含了交换机名、路由键和通过 Snowflake 算法生成的 ID。
设置消息确认回调，当消息发送后，RabbitMQ 会通过回调通知消息是否成功投递。
调用 rabbitTemplate.convertAndSend() 方法向指定的交换机发送消息，同时附带 CorrelationData（包含消息 ID）。
返回生成的消息 ID，以便后续追踪或处理。

关键点：

消息确认机制：使用 ConfirmCallback 确保消息是否成功投递，并且通过日志记录投递状态。
消息 ID：通过 Snowflake 算法生成一个唯一的消息 ID，可以用于追踪和日志记录。
回调机制：确保生产者能在消息投递过程中获得详细的确认信息。

消费者

代码举例

@RabbitHandler@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${rabbitmq.yl.x.queue}", durable = "true"),exchange = @Exchange(value = "${rabbitmq.yl.x.exchange}"),key = "${rabbitmq.yl.x.routing.key}"))public void stateChange(@Payload String payload, @Headers Map<String, Object> headers, Channel channel) throws IOException {log.info("MQ-收到【x】状态变更消息：{} ", payload);final String objDeliveryTag = String.valueOf(headers.get(AmqpHeaders.DELIVERY_TAG));if (NumberUtil.isNumber(objDeliveryTag)) {final long deliveryTag= Long.parseLong(objDeliveryTag);log.info("MQ-收到【x】MQ消息进行消费【执行ack】[{}]", deliveryTag);channel.basicAck(deliveryTag,true);}if(CharSequenceUtil.isNotBlank(payload)) {}}

在这个 Java 方法 leaseStateChange 中，通常是在使用 Spring AMQP 框架与 RabbitMQ 集成时出现的消费消息的处理方法。下面逐个解析这个方法中的各个部分：

方法签名解析

public void leaseStateChange(@Payload String payload, @Headers Map<String, Object> headers, Channel channel) throws IOException

1. `public void leaseStateChange`

public: 表明这个方法是公开的，可以被外部调用。
void: 表明这个方法没有返回值。
leaseStateChange: 这是方法的名称。可以推测这个方法是用来处理与“租赁状态变化”相关的消息的。

2. `@Payload String payload`

@Payload: 这是 Spring AMQP 框架中的一个注解，用于将消息体的内容绑定到方法参数中。在这个例子中，payload 是一个 String 类型的参数，代表从 RabbitMQ 队列中接收到的消息内容。通常，这个消息体是通过 JSON 或其他格式的字符串传递的。
- 注解的作用: @Payload 使得 Spring 能够自动将消息的主体部分注入到方法参数 payload 中。比如，如果消息体是一个 JSON 格式的字符串，Spring 会将其直接赋值给 payload 参数。
- 示例: 假设接收到的消息体是 "{"state": "active", "leaseId": "12345"}"，payload 将会是该字符串。

3. `@Headers Map<String, Object> headers`

@Headers: 这是另一个 Spring AMQP 注解，用来将消息的头部信息注入到方法参数中。RabbitMQ 消息不仅有消息体（payload），还可能包含一些头信息（比如消息的发送时间、路由信息等）。
- 注解的作用: @Headers 会将消息头部的内容绑定到 headers 参数，这个参数是一个 Map<String, Object> 类型，其中键是头部的名称，值是相应的值。头部信息常常用于传递一些附加信息（例如消息的优先级、发送者标识等）。
- 示例: 如果消息头包含如下信息：
  {"correlationId": "abc123", "messageType": "leaseUpdate"}
  那么 headers 将会是一个 Map，其内容是：
  {"correlationId": "abc123", "messageType": "leaseUpdate"}

4. `Channel channel`

Channel: 这是 RabbitMQ 的核心概念之一。Channel 代表一个与 RabbitMQ 服务的连接通道，允许你在该通道上进行消息的消费、确认等操作。
- 作用: 在 Spring AMQP 中，Channel 通常用来进行消息的确认（acknowledge）操作，或者处理消息处理失败时的重新排队等任务。你可以使用 Channel 来手动确认消息，或者控制消息是否成功消费。
- 示例: 如果在消息处理过程中出现异常，消费者可能需要通过 channel.basicNack() 方法来拒绝该消息并可能重新入队。

5. `throws IOException`

throws IOException: 表明这个方法可能会抛出 IOException 异常。RabbitMQ 的消息操作可能会遇到 I/O 错误，因此需要在方法签名中声明可能抛出此异常。通常，这类异常会发生在与 RabbitMQ 的连接中断、消息传输过程失败时等。

Spring AMQP 消费者代码示例

假设这是一个处理来自某个队列的消息的方法，下面是该方法的使用场景和完整代码示例：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.MessageListener;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;import java.io.IOException;
import java.util.Map;@Component
public class LeaseStateChangeListener {// 监听指定队列的消息@RabbitListener(queues = "leaseStateQueue")public void leaseStateChange(@Payload String payload, @Headers Map<String, Object> headers, Channel channel) throws IOException {try {// 处理消息体System.out.println("Received message: " + payload);// 获取消息头部信息String correlationId = (String) headers.get("correlationId");String messageType = (String) headers.get("messageType");System.out.println("CorrelationId: " + correlationId + ", MessageType: " + messageType);// 模拟处理业务逻辑processLeaseStateChange(payload);// 确认消息已成功消费channel.basicAck(headers.hashCode(), false);  // 手动确认消息} catch (Exception e) {// 异常处理，拒绝消息并重新入队System.err.println("Error processing message: " + e.getMessage());channel.basicNack(headers.hashCode(), false, true); // 拒绝并重新入队}}private void processLeaseStateChange(String payload) {// 假设这里是处理租赁状态更新的业务逻辑// 比如将消息解析为对象，进行租赁状态更新System.out.println("Processing lease state change for payload: " + payload);}
}

解析

@RabbitListener: 注解的作用是声明这个方法是一个 RabbitMQ 消息的消费者，并且该方法监听 leaseStateQueue 队列。当有消息到达这个队列时，这个方法会被调用。
消息体 (payload): 这个方法会接收到一个消息体，@Payload 注解将该消息的内容（通常是 JSON 格式的字符串）自动绑定到方法参数 payload 上。
消息头 (headers): 使用 @Headers 注解将消息的头部信息绑定到 headers 参数上，Map<String, Object> 类型。你可以从中获取如 correlationId 和 messageType 等附加信息。
Channel: 这个参数用于消息的确认、拒绝等操作。在成功处理完消息后，调用 channel.basicAck() 来确认消息，表示该消息已经被成功消费。如果处理失败，调用 channel.basicNack() 拒绝该消息，并可以选择是否重新入队。

总结

该方法是一个 RabbitMQ 消费者，用于从指定的队列中消费消息。
通过 @Payload 获取消息体内容，使用 @Headers 获取消息头信息。
使用 Channel 来确认消息的处理状态。
使用 @RabbitListener 注解自动监听队列，并处理消费的消息。

这种方式非常适合处理队列中的业务逻辑，并能够灵活处理消息的确认、拒绝等操作。

消息怎么知道发给哪一个队列

先看队列与交换机怎么绑定的

先创建队列，然后绑定到交换机

RabbitMQ系列-6.如何通过控制台创建交换机、队列、死信队列、延迟队列 - 简书

rabbitMq举例

新来个技术总监，把 RabbitMQ 讲的那叫一个透彻，佩服！ 生产者代码举例 public String sendMsg(final String exchangeName,final String routingKey,final String msg) {} /*** 发送消息* param exchangeName exchangeName* param routin…...

编程日记 2025/6/21 1:28:27

人工智能增强的音频和聊天协作服务

论文标题：AI-enabled Audio and Chat Collaboration Services 中文标题：人工智能增强的音频和聊天协作服务作者信息： Emil P. Andersen, Norwegian Defence Research Establishment (FFI), Kjeller, NorwayJesper R. Goksr, Sindre E. Ha…...

编程日记 2025/6/20 23:06:28

ctr: content digest XXX not found

1、问题描述 ctr 推送镜像到本地仓库报错 ctr: content digest sha256:b96e30ccb0a1e225493e53d4f81cb2c27183406e5f902ed43c4bf15dc6f21c9c: not found 2、问题原因本人的平台是M1，正常情况下载镜像会自动按平台寻找版本，但是由于我下载的镜像版本没…...

编程日记 2025/6/3 18:45:09

java内存模型

文章目录 1.java内存模型2.重排序3.内存屏障3.1四类内存屏障指令 4.happens-before4.1 规则5.volatile5.1特性5.2规则5.3 内存语义5.4 内存屏障插入策略 1.java内存模型 java内存模型（JMM）定义了线程和主内存之间的抽象关系：线程之间的共享变…...

编程日记 2025/6/4 12:26:25

Java Object类与Objects类

1. Object 类 (1) Object类是Java中所有类的祖宗类，因此，Java中所有类的对象都可以直接使用Object类中提供的一些方法(public class Object)。 . (2) Object类常见方法方法说明public String toString()返回对象的字符串形式public boolean eq…...

编程日记 2025/6/3 19:40:15

【目标检查】YOLO系列之：Triton 推理服务器Ultralytics YOLO11

Triton 推理服务器 1、引言2、Triton服务器2.1 什么是Triton Inference Server2.2 将YOLO11 导出为ONNX 格式2.3 设置Triton 模型库2.3.1 创建目录结构2.3.2 将导出的ONNX 模型移至Triton 资源库 2.4 运行Triton 推断服务器2.4.1 使用 Docker 运行Triton Inference Server2.4.2…...

编程日记 2025/6/20 14:14:11

力扣69. x 的平方根

给你一个非负整数 x ，计算并返回 x 的算术平方根。由于返回类型是整数，结果只保留整数部分 ，小数部分将被舍去。注意：不允许使用任何内置指数函数和算符，例如 pow(x, 0.5) 或者 x ** 0.5 。示例 1：…...

编程日记 2025/6/1 3:05:30

关于目标检测YOLO 各版本区别v1-v11/vX/vR

概述 YOLO（You Only Look Once，你只看一次）是一系列开创性的实时目标检测模型，它们彻底改变了计算机视觉领域。由Joseph Redmon开发，后续版本由不同研究人员迭代，YOLO模型以其在图像中检测对象的高速度和准…...

编程日记 2025/6/3 8:34:45

求成绩的平均值，统计低于平均分的人数

【问题描述】编写函数double average(double score[ ], int n,int *m)对存放在实型数组中考试成绩进行求平均分和低于平均分的人数，要求在主函数里面进行班级人数和成绩的输入，调用average函数进行计算成绩平均值和低于平均分的人数后，返回主…...

编程日记 2025/6/21 1:32:33

SGD、BGD、MBGD 之间的区别

文章目录 SGD（Stochastic Gradient Descent，随机梯度下降）BGD（Batch Gradient Descent，批量梯度下降）MBGD（Mini-Batch Gradient Descent，小批量梯度下降） 前置&#xff1…...

编程日记 2025/6/3 10:39:35

Android -- WebView之loadData加载html字符串显示网页

目录前言1. loadUrl 加载网页地址2. loadData 加载Html字符来显示网页3. loadDataWithBaseURL4. 总结前言最近在给一个老项目做64位so文件的适配，当应用发布到应用市场上后，用户反馈64位手机上的网页加载不出内容，但32位的手机上是正常…...

编程日记 2025/6/20 20:42:01

单片机：实现数码管00盗99显示（附带源码）

单片机实现数码管00到99显示在嵌入式开发中，数码管是最常用的显示设备之一，特别适合用于显示数字信息。本项目的目标是利用8051单片机控制数码管实现数字显示，从00到99的循环显示。我们将通过按键或定时器来控制数字的增减，并通…...

编程日记 2025/6/3 20:08:51

大厂面试智力题大全（详细解题思路，持续更新）

目录八个球有一个比较重，问称几次可以把较重的球拎出来，具体称重的方法是怎样的？ 64 匹马 8 个赛道，找出前四名，最少赛多少场（腾讯）一个城市的加油站数量 5L的水桶和3L的水桶怎么量出4L的水有一堆粗细不均匀的绳子，燃烧完一个绳子需要2小时，如何用这些绳子凑出…...

编程日记 2025/6/20 11:10:04

使用 acme.sh 签发和自动续期 ssl https 证书

acme.sh 是一个热度非常高的签发和自动续期 https 证书的工具，虽然官网上提供了充分的操作说明，但是不够简洁，本文以在 nginx 中签发和配置http 为例，列出必要的几个简单步骤。安装因为网络原因，github 大部分人是…...

编程日记 2025/6/19 18:33:52

YOLOv9改进，YOLOv9引入DLKA-Attention可变形大核注意力，WACV2024，二次创新RepNCSPELAN4结构

摘要作者引入了一种称为可变形大核注意力（D-LKA Attention）的新方法来增强医学图像分割。这种方法使用大型卷积内核有效地捕获体积上下文，避免了过多的计算需求。D-LKA Attention 还受益于可变形卷积，以适应不同的数据模式。理论介绍大核卷积（Large Kernel Convolu…...

编程日记 2025/5/29 22:36:47

docker安装mysql8

上一篇提到了docker安装mysql5.7和mysql8有所不同，这一篇说一下mysql8的安装 1、宿主机创建映射目录 mkdir -p /data/mysql8/log mkdir -p /data/mysql8/data mkdir -p /data/mysql8/conf这里我放在了/data/mysql8目录下 2、拉取mysql镜像 docker pull mysql:8.…...

编程日记 2025/6/19 22:55:29

【云原生知识】Kubernets实践-前端服务如何访问后端服务

文章目录概述步骤1：部署后端服务步骤2：配置Nginx步骤3：创建Nginx服务总结如何确保 Nginx 能持续访问后端服务？相关文献概述假设你正在使用Kubernetes作为容器云平台，以下是如何配置Nginx以及相关服务，…...

编程日记 2025/6/19 23:26:33

M3DM的autodl环境构建过程笔记

文章目录在3D-ADS环境https://blog.csdn.net/tfxzgp/article/details/144259472基础上构建(失败的记录，不用看)更换镜像重来（成功）安装缺少的包修改models.py中的RGB和点云backbone的路径修改main.py路径参数运行在3D-ADS环境https://blog.…...

编程日记 2025/5/30 18:04:51

jmeter CLI Mode 传参实现动态设置用户数

一.需求 CLI 运行模式下每次运行想要传入不同的用户数，比如寻找瓶颈值的场景，需要运行多次设置不同的用户数。二.解决思路查看官方API Apache JMeter - Users Manual: Getting Started api CLI Mode 一节中提到可以使用如下参数做属性的替换&#…...

编程日记 2025/6/3 17:43:34

Mac系统下 jdk和maven 安装教程

一、jdk安装教程 1、先去官网选择对应版本下载官网网址：Java SE | Oracle Technology Network | Oracle 中国这里我选择的是jdk8的版本，如果你们想下载更高的版本就选择其他版本，目前大部分公司和教程使用jdk8的版本比较多。点击macos&a…...

编程日记 2025/5/31 21:20:23

pyfink1.20版本下实现消费kafka中数据并实时计算

1、环境 JDK版本：1.8.0_412python版本：3.10.6apache-flink版本：1.20.0flink版本：1.20kafka版本：kafka_2.12-3.1.1flink-sql-connector-kafka版本：3.3.0-1.202、执行python-flink脚本从kafka的demo获取消…...

编程日记 2025/6/3 16:18:54

【经验分享】私有云运维的知识点

最近忙于备考没关注，有次点进某小黄鱼发现首页出现了我的笔记还被人收费了虽然我也卖了一些资源，但我以交流、交换为主，笔记都是免费给别人看的由于当时刚刚接触写的并不成熟，为了避免更多人花没必要的钱，所以决定公…...

编程日记 2025/6/3 14:06:10

影像组学+病理组学+深度学习人工智能应用

影像组学基础学习内容： 特征提取：使用pyradiomics进行形状、纹理、小波变换等特征提取。特征筛选：应用ICC、相关系数、mRMR、Lasso等方法。建模：使用LR、SVM、RF、XGBoost、LightGBM等机器学习算法。模型评估：通过A…...

编程日记 2025/6/20 2:27:52

how to write 述职pptx as a tech manager

As a technical manager, crafting an effective 述职 (performance review) PPT requires you to highlight your leadership, team accomplishments, technical contributions, challenges faced, and future plans. Heres a structured approach to design your PPT: 1. Cov…...

编程日记 2025/6/4 12:19:46

新来个技术总监，把 RabbitMQ 讲的那叫一个透彻，佩服！

生产者

代码举例

分析代码

1. 方法签名

2. CorrelationData 创建与消息 ID 设置

3. 设置回调函数

4. 发送消息

5. 返回消息 ID

总结：

关键点：

消费者

方法签名解析

1. public void leaseStateChange

2. @Payload String payload

3. @Headers Map<String, Object> headers

4. Channel channel

5. throws IOException

Spring AMQP 消费者代码示例

解析

总结

消息怎么知道发给哪一个队列

相关文章：

1. `public void leaseStateChange`

2. `@Payload String payload`

3. `@Headers Map<String, Object> headers`

4. `Channel channel`

5. `throws IOException`