消息队列是后台开发常用的中间件，使用消息队列有下列好处：1、可以使系统异步化，降低响应时间；2、减少不同模块的耦合；3、消除峰值，避免并发请求过多系统无法响应。RabbitMQ 是一个使用 Erlang 开发的消息队列。相比其他消息队列，RabbitMQ 的优势在于文档较全，支持多种客户端。本篇文章简单总结一下 RabbitMQ 的基本概念，使用例子、以及常用的命令行操作。
RabbitMQ 基本概念
- producer： producer 是一个发送消息的应用
- queue： queue 用来存储 producer 发送的消息
- consumer： consumer是接收并处理消息的应用
- exchange：producer 并不会直接将消息发送到 queue 上，而是将消息发送给 exchange，由 exchange 按照一定规则转发给指定队列
其中，exchange 主要有：fanout、direct、topic。在下一部分，就给出几个不同类型的 exchange 的使用实例。
使用例子
fanout
fanout
上图中，P 代表 producer，X 代表 exchange，红色的矩形代表 queue。在 X 的类型为 fanout 的情况下，exchange 将会向所有 queue 广播消息。
下面是 producer 的代码：
import java.io.IOException;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
public class EmitLog {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv)
throws java.io.IOException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明一个 exchange
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String message = getMessage(argv);
//向指定 exchange 发送一个消息。由于是 fanout 模式， routing key 为空
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
channel.close();
connection.close();
}
//...
}
对于 consumer：
import com.rabbitmq.client.*;
import java.io.IOException;
public class ReceiveLogs {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明 exchange
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 生命一个队列，名字随机
String queueName = channel.queueDeclare().getQueue();
//将队列绑定到 exchange 上
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
//consumer 简单的将接收到的消息打印出来
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + message + "'");
}
};
channel.basicConsume(queueName, true, consumer);
}
}
启动多个 consumer 实例，再运行 producer 发送一个消息，可以看到多个 consumer 会同时将消息打印出来。
direct
direct
与 fanout 不同的是，direct 模式下 exchange 在转发消息时会考虑不同队列绑定时指定的 exchange key 是否与消息的 routing key 相同。注意，同一个队列可以绑定多个不同的 exchange key。一个使用场景是将 debug、info、warn、error 等不同级别的日志发送到不同的 queue，分别进行处理：
producer 代码：
public class EmitLogDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv)
throws java.io.IOException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String severity = getSeverity(argv);
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
channel.close();
connection.close();
}
//..
}
consumer 代码：
import com.rabbitmq.client.*;
import java.io.IOException;
public class ReceiveLogsDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String queueName = channel.queueDeclare().getQueue();
if (argv.length < 1){
System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
System.exit(1);
}
for(String severity : argv){
channel.queueBind(queueName, EXCHANGE_NAME, severity);
}
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
}
};
channel.basicConsume(queueName, true, consumer);
}
}
上面的代码中，producer 在发送时指定了日志级别，而 consumer 则将自己的 queue 绑定在命令行传入的的若干 exchange key 上。可以看到，只有绑定了对应的 exchange key，consumer 才能接收到 producer 发来的信息。
topic
topic exchange 与 direct exchange 类似，不过 topic exchange 的 routing key 必须是由 . 分割，并且 topic exchange 支持队列绑定exchange key 时使用通配符。可以使用的通配符有:
-
*可以表示任何单独的单词 -
#可以表示0个或任意多个单词.
topic
比如上图，quick.orange.rabbit 会被转发给Q1和Q2，同样，'lazy.orange.elephant' 也会被同时转发给Q1和Q2。当队列绑定exchange 时的 exchange key 不包含通配符时，topic exchange 其实与 direct exchange 相同。而 exchange key 为 # 时，topic exchange 又可以看做一个 fanout exchange。
producer 示例：
public class EmitLogTopic {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv)
throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String routingKey = getRouting(argv);
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
connection.close();
}
//...
}
consumer 示例：
import com.rabbitmq.client.*;
import java.io.IOException;
public class ReceiveLogsTopic {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String queueName = channel.queueDeclare().getQueue();
if (argv.length < 1) {
System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
System.exit(1);
}
for (String bindingKey : argv) {
channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
}
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
}
};
channel.basicConsume(queueName, true, consumer);
}
}
消息的持久化
一般情况下，在 RabbitMQ crash 或者是被人为关闭后，队列和队列中的消息都将消失。如果需要消息持久化，需要进行如下配置：
- 告诉rabbitMQ 需要将队列持久化：
```java
boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);
```
- 发送消息时在消息的 property 中设置消息持久化
```java
channel.basicPublish("", "task_queue",
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes());
```
不过这种持久化还不能保证消息完全不丢失。虽然这种方式会告诉 RabbitMQ 将消息存储到磁盘中，但是在收到消息到存储到磁盘之间还有一段时间窗口。其次，RabbitMQ 并不会每次都进行 fsync 操作，这样消息其实是批量被存储到磁盘上的。如果需要确保每个消息都不丢失，可以使用 publisher confirm。
autoAck
consumer 在消费消息时，可以选择是否进行 autoAck。如果 autoAck 为true，一旦消息被投递给 consumer 后，就会被直接删除。这样如果在处理消息时 consumer 挂了，这个消息就永远丢失了。通过关闭 autoAck，我们可以在完成逻辑后自己对消息进行 ack:
channel.basicQos(1); // accept only one unack-ed message at a time (see below)
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + message + "'");
try {
doWork(message);
} finally {
System.out.println(" [x] Done");
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);
常用cmdline 命令
- 启动 RabbitMQ 并从shell detach
sudo rabbitmq-server -detached
- 列出所有 exchange 或 binding
sudo rabbitmqctl list_exchanges
rabbitmqctl list_queues
- 添加一个用户
rabbitmqctl add_user Username Password
- 展示所有用户
rabbitmqctl list_users
- 删除用户
rabbitmqctl delete_user Username
- 设置用户tag
rabbitmqctl set_user_tags User Tag
- 设置用户权限 (RabbitMQ有虚拟主机的概念呢，默认的vhost 为 /)
rabbitmqctl set_permissions -p /vhost Usernames '.*' '.*' '.*'
set_permissions [-p <vhost>] <user> <conf> <write> <read>
其他一些命令可以查看官网文档。
使用 RabbitMQ 实现延时消息
延迟消息有很多使用场景，比如收货后24小时自动给订单好评、创建订单4小时未支付将订单关闭并返回库存等。实现延时消息有很多种思路，这里介绍如何使用 RabbitMQ 的 Dead letter exchange 来简单的实现延时消息。
下面是的代码整合 spring 与 rabbitMQ，通过 Java Config 来进行配置：
//创建 connectionFactory
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory =
new CachingConnectionFactory(host, Integer.parseInt(port));
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
return connectionFactory;
}
// RabbitAdmin
@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(connectionFactory());
}
//接收并处理消息的队列
@Bean
Queue queue() {
return QueueBuilder.durable(queueName)
.build();
}
//创建一个topicExchange
@Bean
TopicExchange exchange() {
return new TopicExchange(exchange);
}
// 使用exchangeKey将队列绑定到exchange上
@Bean
Binding binding() {
return BindingBuilder.bind(queue()).to(exchange()).with(exchangeKey);
}
//用户创建 MessageDispatcher 用于处理消息
@Bean
MessageDispatcher messageDispatcher() {
return new RabbitmqMessageDispatcher();
}
//配置消息将由RabbitmqMessageDispatcher的dispatch方法处理
@Bean
MessageListenerAdapter listenerAdapter() {
return new MessageListenerAdapter(messageDispatcher(), "dispatch");
}
//设置消息处理失败时的重试策略
@Bean
public StatefulRetryOperationsInterceptor statefulRetryOperationsInterceptor() {
return RetryInterceptorBuilder.stateful()
.maxAttempts(retryNum)
.backOffOptions(initRetryInterval,
retryMultiplier,
retryMaxInterval) // initialInterval, multiplier, maxInterval
.build();
}
//避免由于客户端没有带messageId导致receiver出现问题
@Bean
public MissingMessageIdAdvice missingIdAdvice() {
return new MissingMessageIdAdvice(new MapRetryContextCache());
}
// 进行消息转换 这里是将byte[]转为String
@Bean
MessageConverter simpleMessageConverter() {
SimpleMessageConverter messageConverter = new SimpleMessageConverter();
messageConverter.setCreateMessageIds(true);
return messageConverter;
}
//配置consumer以及对应的队列
@Bean
SimpleMessageListenerContainer container() {
SimpleMessageListenerContainer container
= new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueueNames(queueName);
container.setMessageListener(listenerAdapter());
container.setAdviceChain(
new Advice[] {missingIdAdvice(), statefulRetryOperationsInterceptor()});
container.setMessageConverter(simpleMessageConverter());
return container;
}
//延时队列 注意配置 x-dead-letter-exchange x-dead-letter-routing-key
@Bean
Queue delayQueue() {
return QueueBuilder.
durable(delayQueueName).
autoDelete().
withArgument("x-dead-letter-exchange", exchange).
withArgument("x-dead-letter-routing-key", exchangeKey).
build();
}
//创建延时exchange
@Bean
TopicExchange delayExchange() {
return new TopicExchange(delayExchange);
}
//将延时队列与延时 exchangeKey 绑定
@Bean
Binding delayQueueBinding() {
return BindingBuilder.
bind(delayQueue()).
to(delayExchange()).
with(delayExchangeKey);
}
这样，producer 发送消息时设置消息的 expiration 属性，然后将消息发送到延时exchange，而后在 expiration 时间后消息将被转发到x-dead-letter-exchange 指定的 exchange 去，而后被正常消费，实现了延时消息功能。
