[心得体会]RabbitMQ

• RabbitMQ是什么? 

AMQP，即Advanced Message Queuing Protocol，
一个提供统一消息服务的应用层标准高级消息队列协议，
是应用层协议的一个开放标准，为面向消息的中间件设计。
基于此协议的客户端与消息中间件可传递消息，
并不受客户端/中间件不同产品，不同的开发语言等条件的限制。

AMQP，即Advanced Message Queuing Protocol，

一个提供统一消息服务的应用层标准高级消息队列协议，

是应用层协议的一个开放标准，为面向消息的中间件设计。

基于此协议的客户端与消息中间件可传递消息，

并不受客户端/中间件不同产品，不同的开发语言等条件的限制。

• RabbitMQ安装
  

• Linux安装

rpm -Uvh http://www.rabbitmq.com/releases/erlang/erlang-18.1-1.el7.centos.x86_64.rpm

rpm -Uvh http://www.rabbitmq.com/releases/erlang/erlang-18.1-1.el7.centos.x86_64.rpm

rpm -Uvh http://www.rabbitmq.com/releases/rabbitmq-server/v3.5.6/rabbitmq-server-3.5.6-1.noarch.rpm

rpm -Uvh http://www.rabbitmq.com/releases/rabbitmq-server/v3.5.6/rabbitmq-server-3.5.6-1.noarch.rpm

rpm -qa|grep rabbitmq

rpm -qa|grep rabbitmq

$ sudo chkconfig rabbitmq-server on  # 添加开机启动RabbitMQ服务
$ sudo /sbin/service rabbitmq-server start # 启动服务
$ sudo /sbin/service rabbitmq-server status  # 查看服务状态
$ sudo /sbin/service rabbitmq-server stop   # 停止服务

# 查看当前所有用户
$ sudo rabbitmqctl list_users

# 查看默认guest用户的权限
$ sudo rabbitmqctl list_user_permissions guest

# 由于RabbitMQ默认的账号用户名和密码都是guest。为了安全起见, 先删掉默认用户
$ sudo rabbitmqctl delete_user guest

# 添加新用户
$ sudo rabbitmqctl add_user username password

# 设置用户tag
$ sudo rabbitmqctl set_user_tags username administrator

# 赋予用户默认vhost的全部操作权限
$ sudo rabbitmqctl set_permissions -p / username ".*" ".*" ".*"

# 查看用户的权限
$ sudo rabbitmqctl list_user_permissions username

$ sudo chkconfig rabbitmq-server on  # 添加开机启动RabbitMQ服务

$ sudo /sbin/service rabbitmq-server start # 启动服务

$ sudo /sbin/service rabbitmq-server status  # 查看服务状态

$ sudo /sbin/service rabbitmq-server stop   # 停止服务

# 查看当前所有用户

$ sudo rabbitmqctl list_users

# 查看默认guest用户的权限

$ sudo rabbitmqctl list_user_permissions guest

# 由于RabbitMQ默认的账号用户名和密码都是guest。为了安全起见, 先删掉默认用户

$ sudo rabbitmqctl delete_user guest

# 添加新用户

$ sudo rabbitmqctl add_user username password

# 设置用户tag

$ sudo rabbitmqctl set_user_tags username administrator

# 赋予用户默认vhost的全部操作权限

$ sudo rabbitmqctl set_permissions -p / username ".*" ".*" ".*"

# 查看用户的权限

$ sudo rabbitmqctl list_user_permissions username

开启web管理接口

如果只从命令行操作RabbitMQ，多少有点不方便。幸好RabbitMQ自带了web管理界面，只需要启动插件便可以使用。

$ sudo rabbitmq-plugins enable rabbitmq_management

$ sudo rabbitmq-plugins enable rabbitmq_management

配置RabbitMQ

关于RabbitMQ的配置，可以下载RabbitMQ的配置文件模板到/etc/rabbitmq/rabbitmq.config, 然后按照需求更改即可。
关于每个配置项的具体作用，可以参考官方文档。
更新配置后，别忘了重启服务哦！

开启用户远程访问

默认情况下，RabbitMQ的默认的guest用户只允许本机访问， 如果想让guest用户能够远程访问的话，只需要将配置文件中的loopback_users列表置为空即可，如下：

{loopback_users, []}

{loopback_users, []}

{loopback_users, ["admin"]}

{loopback_users, ["admin"]}

channel.basicQos(1); // accept only one unack-ed message at a time (see below)

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  String message = new String(delivery.getBody(), "UTF-8");

  System.out.println(" [x] Received ‘" + message + "‘");
  try {
    doWork(message);
  } finally {
    System.out.println(" [x] Done");
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
  }
};
boolean autoAck = false; // 手动Ack功能开始
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });

rabbitmqctl.bat list_queues name messages_ready messages_unacknowledged

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);

int prefetchCount = 1;
channel.basicQos(prefetchCount);

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

public class NewTask {

  private static final String TASK_QUEUE_NAME = "task_queue";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    try (Connection connection = factory.newConnection();
         Channel channel = connection.createChannel()) {
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

        String message = String.join(" ", argv);

        channel.basicPublish("", TASK_QUEUE_NAME,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                message.getBytes("UTF-8"));
        System.out.println(" [x] Sent ‘" + message + "‘");
    }
  }

}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class Worker {

  private static final String TASK_QUEUE_NAME = "task_queue";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    final Connection connection = factory.newConnection();
    final Channel channel = connection.createChannel();

    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

    channel.basicQos(1);

    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), "UTF-8");

        System.out.println(" [x] Received ‘" + message + "‘");
        try {
            doWork(message);
        } finally {
            System.out.println(" [x] Done");
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    };
    channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { });
  }

  private static void doWork(String task) {
    for (char ch : task.toCharArray()) {
        if (ch == ‘.‘) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException _ignored) {
                Thread.currentThread().interrupt();
            }
        }
    }
  }
}

* 为消息添加TTL时间的方式有两种
* (1) 为添加进入队列的每个消息都加上TTL
* (2) 在消息即将公布的时候, 给消息添加上消息的ttl

(1) 为添加进入队列的每个消息都加上TTL

Map<String, Object> arg = new HashMap<>();
arg.put("x-message-ttl", 60000);
channel.queueDeclare(QUEUE, true, false, false, arg);

(2) 在消息即将公布的时候, 给消息添加上消息的ttl

• 同一个消息在不同消息队列中是否会同时寿终正寝?

• 如何修改TTL的时间? 

rabbitmqctl set_policy TTL ".*" ‘{"message-ttl":60000}‘ --apply-to queues

rabbitmqctl set_policy TTL ".*" "{""message-ttl"":60000}" --apply-to queues

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 60000);
channel.queueDeclare("myqueue", false, false, false, args);

• 在消息即将公布的时候, 给消息添加上ttl时间

rabbitmqctl set_policy expiry ".*" ‘{"expires":1800000}‘ --apply-to queues

rabbitmqctl.bat set_policy expiry ".*" "{""expires"":1800000}" --apply-to queues

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-expires", 1800000);
channel.queueDeclare("myqueue", false, false, false, args);

// 为添加进入队列的每个消息都加上TTL
// Map<String, Object> arg = new HashMap<>();
// arg.put("x-message-ttl", 60000);
// 为队列添加TTL , 半个小时的时间
Map<String, Object> arg = new HashMap<>();
arg.put("x-expires", 1800000);
channel.queueDeclare(QUEUE, true, false, false, arg);
String message = "Hello World";
// 在消息即将公布的时候, 给消息添加上消息的ttl
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("600000").build();
channel.basicPublish("my-exchange", QUEUE, properties, message.getBytes());

package com.xuecheng.manage_cms.rabbit;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

/**
 * @author zhengwei
 * @version 1.0.0
 * @date 2019/12/4 20:29
 * @msg
 **/
public class Producer01 {
   
   private final static String QUEUE = "helloworld";
   
   public static void main(String[] args) throws Exception {
      Connection connection = null;
      Channel channel = null;
      try {
         ConnectionFactory factory = new ConnectionFactory();
         factory.setHost("localhost");
         connection = factory.newConnection();
         channel = connection.createChannel();
         channel.queueDeclare(QUEUE, true, false, false, null);
         String message = "Hello World";
         // 将生产者的通道设置为持久化状态
         channel.basicPublish("", QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
         System.out.println("[x] Sent‘" + message + "‘");
      } catch (Exception e) {
          e.printStackTrace();
      } finally {
         if (channel != null) {
            channel.close();
         }
         if (connection != null) {
             connection.close();
         }
      }
   }

}

package com.xuecheng.manage_cms.rabbit;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @author zhengwei
 * @version 1.0.0
 * @date 2019/12/4 20:50
 * @msg
 **/
public class Consumer01 {
   
   private static final String QUEUE = "helloworld";
   
   public static void main(String[] args) throws Exception {
      func1();
   }
   
   public static void func1() throws Exception {
      ConnectionFactory factory = new ConnectionFactory();
      factory.setHost("localhost");
      Connection connection = factory.newConnection();
      Channel channel = connection.createChannel();
      channel.queueDeclare(QUEUE, true, false, false, null);
      channel.basicQos(1);
      DeliverCallback deliverCallback = (consumerTag, delivery) -> {
         String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
         System.out.println("receive message..." + message);
         try {
            doWork(message);
         } catch (InterruptedException e) {
            e.printStackTrace();
         } finally {
            System.out.println(" [x] Done");
            // 应答
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
         }
      };
      // 开启消费者的自动 Ack 功能
      boolean autoAck = false;
      channel.basicConsume(QUEUE, autoAck, deliverCallback, (consumerTag) -> {
         System.out.println("consumerTag: " + consumerTag);
      });
   }
   
   public static void doWork(String task) throws InterruptedException {
      for (char ch : task.toCharArray()) {
         if (ch == ‘.‘) {
            Thread.sleep(1000);
         }
      }
   }
   
   public static void func() throws Exception {
      ConnectionFactory factory = new ConnectionFactory();
      factory.setHost("localhost");
      Connection connection = factory.newConnection();
      Channel channel = connection.createChannel();
      channel.queueDeclare(QUEUE, true, false, false, null);
      DefaultConsumer consumer = new DefaultConsumer(channel) {
         @Override
         public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            String exchange = envelope.getExchange();
            String routingKey = envelope.getRoutingKey();
            long deliveryTag = envelope.getDeliveryTag();
            String msg = new String(body, "utf-8");
            System.out.println("receive message..." + msg);
            channel.basicAck(envelope.getDeliveryTag(), false);
            //true if the rejected message should be requeued rather than discarded/dead-lettered
            // 如果拒绝的方式是 true 则这个消息不被丢弃, false的话直接丢弃消息
            channel.basicReject(envelope.getDeliveryTag(), true);
         }
      };
      channel.basicConsume(QUEUE, true, consumer);
   }
}

sudo rabbitmqctl list_exchanges

channel.basicPublish( "logs", "", null, message.getBytes());

channel.queueBind(queueName, "logs", "");

public class EmitLog {

   private static final String EXCHANGE_NAME = "logs";
   
   public static void main(String[] args) throws Exception {
      ConnectionFactory factory = new ConnectionFactory();
      Connection connection = null;
      Channel channel = null;
      try {
         connection = factory.newConnection();
         channel = connection.createChannel();
         // 声明一个扇形交换机, 向所有的消费者发布消息
         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
         String message = "zhazha Hello World";
         // 指定 logs(EXCHANGE_NAME) 交换器, 将消息发布出去
         channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes(StandardCharsets.UTF_8));
      } catch (Exception e) {
          e.printStackTrace();
      } finally {
         if (channel != null) {
            channel.close();
         }
         if (connection != null) {
            connection.close();
         }
      }
   }
   
}

public class ReceiveLogs {
   
   private static final String EXCHANGE_NAME = "logs";
   
   public static void main(String[] args) throws Exception {
      ConnectionFactory factory = new ConnectionFactory();
      Connection connection = factory.newConnection();
      Channel channel = connection.createChannel();
      String queueName = channel.queueDeclare().getQueue();
      channel.queueBind(queueName, EXCHANGE_NAME, "");
      DeliverCallback deliverCallback = (consumerTag, message) -> {
         System.out.println("message = " + new String(message.getBody(), StandardCharsets.UTF_8));
      };
      channel.basicConsume(queueName, true, deliverCallback, (consumerTag) -> {
      });
   }
   
}

public class EmitLogDirect {
   
   private static final String EXCHANGE_NAME = "direct_logs";
   
   public static void main(String[] args) throws Exception {
      ConnectionFactory factory = new ConnectionFactory();
      factory.setHost("localhost");
      Connection connection = factory.newConnection();
      Channel channel = connection.createChannel();
      channel.exchangeDeclare(EXCHANGE_NAME, "direct");
      String routingKey = "info";
      String message = "zhazha";
      channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes(StandardCharsets.UTF_8));
      System.out.println(" [x] Sent ‘" + routingKey + "‘:‘" + message + "‘");
   }

}

public class ReceiveLogsDirect {
   
   private static final String EXCHANGE_NAME = "direct_logs";
   
   public static void main(String[] args) throws IOException, TimeoutException {
      String[] argv = new String[]{"info", "warning", "error"};
      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();
      for (String routingKey : argv) {
         channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
      }
      System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
      DeliverCallback deliverCallback = (consumerTag, delivery) -> {
         String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
         System.out.println(" [x] Received ‘" + delivery.getEnvelope().getRoutingKey() + "‘:‘" + message + "‘");
      };
      channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
      });
   }
   
}