社区论坛自助建站网江阴网站优化公司
概述
RabbitTemplate 是 Spring AMQP（Advanced Message Queuing Protocol）提供的核心类之一，它是 AmqpTemplate 接口的主要实现，用于与 RabbitMQ 交互。
它封装了消息的生产和消费逻辑，使得开发者可以更方便地与 RabbitMQ 进行通信。
一、RabbitTemplate 主要功能
- 发送消息到指定的交换机和队列
- 从队列中接收消息
- 发送并接收消息，实现 RPC（远程过程调用）
- 支持事务管理
- 提供消息转换器，支持 JSON、Java 序列化等格式
- 处理消息确认、失败回调等
二、RabbitTemplate 常用方法
1. 消息发送
void convertAndSend(String exchange, String routingKey, Object message);
void convertAndSend(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor);
void convertAndSend(String routingKey, Object message);
void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor);
-
参数说明：
exchange：交换机名称routingKey：路由键message：要发送的消息（可以是String、Map、对象等）messagePostProcessor：消息后处理器，可用于修改消息的属性
-
示例：
rabbitTemplate.convertAndSend("my-exchange", "my-routing-key", "Hello RabbitMQ!");
2. 消息接收
Object receiveAndConvert(String queueName);
Message receive(String queueName);
-
参数说明：
queueName：队列名称
-
返回值：
receiveAndConvert：自动将消息转换为 Java 对象receive：返回原始Message对象，包含body和properties
-
示例：
String message = (String) rabbitTemplate.receiveAndConvert("my-queue");
System.out.println("接收到消息：" + message);
3. 发送并接收消息（RPC 模式）
Object convertSendAndReceive(String exchange, String routingKey, Object message);
-
应用场景：
- 客户端发送消息到
exchange - 服务器处理后返回响应
- 客户端接收处理结果
- 客户端发送消息到
-
示例：
String response = (String) rabbitTemplate.convertSendAndReceive("my-exchange", "my-routing-key", "Hello");
System.out.println("收到的响应：" + response);
4. 设置消息确认回调
- 配置
ConfirmCallback确认消息是否到达交换机
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {if (ack) {System.out.println("消息成功到达交换机");} else {System.out.println("消息未到达交换机，原因：" + cause);}
});
- 配置
ReturnCallback确认消息是否到达队列
rabbitTemplate.setMandatory(true); // 必须设置为 true，否则 returnCallback 不生效
rabbitTemplate.setReturnsCallback(returned -> {System.out.println("消息未到达队列：" + returned.getReplyText());
});
5. 事务管理
- 开启事务：
rabbitTemplate.setChannelTransacted(true);
- 事务提交：
rabbitTemplate.convertAndSend("exchange", "key", "message");
rabbitTemplate.execute(channel -> {channel.txCommit();return null;
});- 事务回滚：
rabbitTemplate.execute(channel -> {channel.txRollback();return null;
});
三、应用场景
-
简单的生产者-消费者模型
RabbitTemplate用于发送消息到队列，消费者监听队列处理消息。
-
发布/订阅模式
RabbitTemplate发送消息到fanout交换机，多个队列接收相同的消息。
-
RPC 远程调用
- 发送请求消息到队列，并等待响应结果。
-
事务管理
- 确保消息在 RabbitMQ 处理时的可靠性。
四、完整示例代码
1. 生产者
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;@Service
public class RabbitMqProducer {private final RabbitTemplate rabbitTemplate;public RabbitMqProducer(RabbitTemplate rabbitTemplate) {this.rabbitTemplate = rabbitTemplate;}public void sendMessage(String message) {rabbitTemplate.convertAndSend("my-exchange", "my-routing-key", message);System.out.println("发送消息：" + message);}
}
2. 消费者
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;@Service
public class RabbitMqConsumer {@RabbitListener(queues = "my-queue")public void receiveMessage(String message) {System.out.println("收到消息：" + message);}
}
3. 配置类
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMqConfig {@Beanpublic Exchange exchange() {return new DirectExchange("my-exchange");}@Beanpublic Queue queue() {return new Queue("my-queue");}@Beanpublic Binding binding(Queue queue, Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("my-routing-key").noargs();}
}
五、总结
| 方法 | 作用 |
|---|---|
convertAndSend() | 发送消息 |
receiveAndConvert() | 接收并转换消息 |
receive() | 仅接收消息（不转换） |
convertSendAndReceive() | 发送消息并等待响应 |
setConfirmCallback() | 监听消息是否到达交换机 |
setReturnsCallback() | 监听消息是否到达队列 |
setChannelTransacted(true) | 开启事务支持 |
RabbitTemplate 适用于大多数 RabbitMQ 操作，结合 Spring Boot 配置可以大大简化开发。