Rabbitmq 使用小记

作者: millions_chan | 来源:发表于2017-04-02 22:47 被阅读4032次

消息队列是后台开发常用的中间件，使用消息队列有下列好处：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

上图中，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

与 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 去，而后被正常消费，实现了延时消息功能。

Rabbitmq 使用小记

RabbitMQ 基本概念

使用例子

fanout

direct

topic

消息的持久化

autoAck

常用cmdline 命令

使用 RabbitMQ 实现延时消息

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读