新来个技术总监，把 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.如何通过控制台创建交换机、队列、死信队列、延迟队列 - 简书
