RabbitTemplate类介绍、应用场景和示例代码

概述

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;
});

三、应用场景

1. 简单的生产者-消费者模型

   • RabbitTemplate 用于发送消息到队列，消费者监听队列处理消息。

2. 发布/订阅模式

   • RabbitTemplate 发送消息到 fanout 交换机，多个队列接收相同的消息。

3. RPC 远程调用

   • 发送请求消息到队列，并等待响应结果。

4. 事务管理

   • 确保消息在 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 {

    @Bean
    public Exchange exchange() {
        return new DirectExchange("my-exchange");
    }

    @Bean
    public Queue queue() {
        return new Queue("my-queue");
    }

    @Bean
    public Binding binding(Queue queue, Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("my-routing-key").noargs();
    }
}

五、总结

RabbitTemplate 适用于大多数 RabbitMQ 操作，结合 Spring Boot 配置可以大大简化开发。