引言

RabbitMQ是最流行的消息队列之一。本文介绍RabbitMQ的基础概念和使用方法。

一、核心概念

1
Producer(生产者) → Exchange(交换机) → Queue(队列) → Consumer(消费者)

1.1 主要组件

  • Producer：发送消息
  • Exchange：路由消息
  • Queue：存储消息
  • Consumer：接收消息
  • Binding：绑定Exchange和Queue

二、Exchange类型

2.1 Direct Exchange（直连）

1
2
3
4
5
6
// 生产者
channel.exchangeDeclare("direct_logs", "direct");
channel.basicPublish("direct_logs", "error", null, message.getBytes());

// 消费者
channel.queueBind(queueName, "direct_logs", "error");

2.2 Fanout Exchange（广播）

1
2
3
// 发送到所有绑定的队列
channel.exchangeDeclare("logs", "fanout");
channel.basicPublish("logs", "", null, message.getBytes());

2.3 Topic Exchange（主题）

1
2
3
4
5
6
// 支持通配符路由
channel.exchangeDeclare("topic_logs", "topic");
channel.basicPublish("topic_logs", "user.create", null, message.getBytes());

// 消费者
channel.queueBind(queueName, "topic_logs", "user.*");

2.4 Headers Exchange（头部）

1
2
3
4
// 根据消息头部路由
Map<String, Object> headers = new HashMap<>();
headers.put("format", "pdf");
channel.basicPublish("headers_exchange", "", props, message.getBytes());

三、工作模式

3.1 简单队列

1
2
3
4
5
6
7
// 生产者
channel.queueDeclare("hello", false, false, false, null);
channel.basicPublish("", "hello", null, message.getBytes());

// 消费者
channel.queueDeclare("hello", false, false, false, null);
channel.basicConsume("hello", true, deliverCallback, consumerTag -> {});

3.2 工作队列

1
2
3
// 多个消费者竞争消费
channel.basicQos(1); // 每次只处理一条
channel.basicConsume("task_queue", false, deliverCallback, consumerTag -> {});

3.3 发布/订阅

1
2
3
4
// Exchange广播到多个队列
channel.exchangeDeclare("logs", "fanout");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, "logs", "");

四、消息确认

4.1 消费者确认

1
2
3
4
5
6
7
8
9
10
// 手动确认
channel.basicConsume(QUEUE_NAME, false, (consumerTag, delivery) -> {
    try {
        // 处理消息
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 拒绝消息
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

4.2 生产者确认

1
2
3
4
5
6
7
8
9
// 开启确认模式
channel.confirmSelect();
channel.basicPublish("", "queue_name", null, message.getBytes());

if (channel.waitForConfirms()) {
    System.out.println("消息已确认");
} else {
    System.out.println("消息发送失败");
}

五、Spring Boot集成

5.1 配置

1
2
3
4
5
6
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

5.2 生产者

1
2
3
4
5
6
@Autowired
private RabbitTemplate rabbitTemplate;

public void sendMessage(String message) {
    rabbitTemplate.convertAndSend("exchange", "routing.key", message);
}

5.3 消费者

1
2
3
4
@RabbitListener(queues = "queue_name")
public void receiveMessage(String message) {
    System.out.println("收到消息: " + message);
}

六、最佳实践

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
1. 使用消息持久化
channel.queueDeclare("queue", true, false, false, null);
MessageProperties.PERSISTENT_TEXT_PLAIN

2. 设置预取数量
channel.basicQos(1);

3. 使用死信队列
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx_exchange");

4. 设置消息过期时间
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .expiration("60000")
    .build();

总结

RabbitMQ核心：

  1. Exchange类型：Direct、Fanout、Topic
  2. 工作模式：简单队列、工作队列、发布/订阅
  3. 消息确认：保证消息可靠性
  4. Spring集成：简化开发

参考资料