0. 概述

0.1 什么是MQ？

Message Queue（MQ），消息队列中间件。消息队列就是在消息传输过程中保存消息的容器。

0.2 为什么要使用MQ？

• 提高系统响应速度
• 提高系统稳定性
• 降低系统耦合性

0.3 RabbitMQ概念

• 生成者、消费者

Producer：消息的生产者
Consumer：消息的消费者

• Queue
  消息队列，提供了 FIFO 的处理机制，具有缓存消息的能力。RabbitMQ中，队列消息可以设置为持久化，临时或者自动删除。

• ExChange
  Exchange 类似于数据通信网络中的交换机，提供消息路由策略。

RabbitMQ 中，Producer 不是通过信道直接将消息发送给 Queue，而是先发送给 ExChange。一个 ExChange 可以和多个 Queue 进行绑定，Producer 在传递消息的时候，会传递一个 ROUTING_KEY，ExChange 会根据这个 ROUTING_KEY 按照特定的路由算法，将消息路由给指定的 Queue。

0.4 Exchange类型

• direct（默认）：直接交换器，工作方式类似于单播，ExChange 会将消息发送完全匹配 ROUTING_KEY 的 Queue（key 就等于 queue）
  

  direct.jpeg

• fanout：广播是式交换器，不管消息的 ROUTING_KEY 设置为什么，ExChange 都会将消息转发给所有绑定的 Queue（无视 key，给所有的 queue 都来一份）
  

  fanout.jpeg

• topic：主题交换器，工作方式类似于组播，ExChange 会将消息转发和 ROUTING_KEY 匹配模式相同的所有队列（key 可以用“宽字符”模糊匹配 queue），比如，ROUTING_KEY 为 user.stock 的 Message 会转发给绑定匹配模式为 * .stock,user.stock， * . * 和 #.user.stock.# 的队列。（ * 表是匹配一个任意词组，# 表示匹配 0 个或多个词组）

  
  
  topic.jpeg

• headers(不推荐)：消息体的 header 匹配，无视 key，通过查看消息的头部元数据来决定发给那个 queue（AMQP 头部元数据非常丰富而且可以自定义）。headers 类型的交换器性能会很差，而且也不实用，基本上不会看到它的存在

1. RabbitMQ安装及使用

Rabbit官方文档：https://www.rabbitmq.com/documentation.html

1.1 使用Docker安装RabbitMQ

Rabbit是基于Erlang开发的，因此直接安装需要先安装Erlang环境，比较繁琐（具体安装步骤可以看官方文档）
这里建议使用Docker安装

# 安装命令
docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

# docker-compose
version: '3.1'
services:
  rabbitmq:
    restart: always
    image: rabbitmq:management
    container_name: rabbitmq
    ports:
      - 5672:5672
      - 15672:15672
    environment:
      TZ: Asia/Shanghai
      RABBITMQ_DEFAULT_USER: rabbit
      RABBITMQ_DEFAULT_PASS: 123456
    volumes:
      - ./data:/var/lib/rabbitmq

# 启动后访问：http://ip:15627

1.2 Topic exchange 实例（Java）

下面是安装官方教程整理的实例：
https://www.rabbitmq.com/tutorials/tutorial-five-java.html

• 导入依赖

# pom.xml 依赖
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.7.1</version>
</dependency>

• 生产者客户端

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

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");
    factory.setUsername("rabbit");
    factory.setPassword("123456");
    
    try (Connection connection = factory.newConnection();
         Channel channel = connection.createChannel()) {
        
        // 声明 exchange 类型为Topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        String routingKey = getRouting(argv);
        String message = getMessage(argv);

        // 向exchange发送带有routingKey的消息
        channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
    }
  }
  //..
}

• 消费者客户端

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

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");
    factory.setUsername("rabbit");
    factory.setPassword("123456");
    
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    // 声明 exchange 类型为Topic
    channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    // 创建一个临时queue
    String queueName = channel.queueDeclare().getQueue();

    if (argv.length < 1) {
        System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
        System.exit(1);
    }

    for (String bindingKey : argv) {
        // 使用bindingKey绑定exchange和queue
        channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
    }

    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

    // 定义消息回调
    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), "UTF-8");
        System.out.println(" [x] Received '" +
            delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
    };
    channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
  }
}

• 运行结果

# 订阅 logs.* 消息日志
java -cp $CP ReceiveLogsTopic "logs.*"

# 订阅 logs.error 消息日志
java -cp $CP ReceiveLogsTopic "logs.error"

# 发送日志
java -cp $CP EmitLogTopic "logs.error" "Runtime error"  

# 两个订阅者都能收到 "Runtime error" 消息

1.3 Spring-Rabbit

• 导入依赖

选择连接器:
PooledChannelConnectionFactory：
ThreadChannelConnectionFactory
CachingConnectionFactory