RabbitMQ详解

董镇进

0. 概述   

    MQ，中文是消息队列（MessageQueue），字面来看就是存放消息的队列。也就是事件驱动架构中的Broker，从而实现异步调用(同步调用用Feign)


比较常见的MQ实现：

• ActiveMQ
  
• RabbitMQ
  
• RocketMQ
  
• Kafka
  

1.RabbitMQ安装和概述

1.1 安装

直接用Docker安装：
docker run \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
-v mq-plugins:/plugins \
--name mq \
--hostname mq \
-p 15672:15672 \
-p 5672:5672 \
-d \
rabbitmq:3.8-management1.2 概述



RabbitMQ中的一些角色：

• publisher：生产者
  
• consumer：消费者
  
• exchange：交换机，负责消息路由
  
• queue：队列，存储消息
  
• virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离
  

2. RabbitMQ的常用消息模型



3. 入门案例

包括三部分：

• mq-demo：父工程，管理项目依赖
  
• publisher：消息的发送者
  
• consumer：消息的消费者
  

1.Publisher
public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("admin");
        factory.setPassword("xxxx");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功：【" + message + "】");

        // 5.关闭通道和连接
        channel.close();
        connection.close();

    }
}

2. customer
public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("admin");
        factory.setPassword("xxxx");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息：【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}4. SpringAMQP



SpringAMQP提供了三个功能：

• 自动声明队列、交换机及其绑定关系
  
• 基于注解的监听器模式，异步接收消息
  
• 封装了RabbitTemplate工具，用于发送消息
  

4.1 Basic Queue



流程如下：

• 在父工程中引入spring-amqp的依赖
  
• 在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列
  
• 在consumer服务中编写消费逻辑，绑定simple.queue这个队列
  

1.导入依赖
    <artifactId>spring-boot-starter-amqp</artifactId>

2. publisher模块中搞事情
2.1 首先配置MQ地址，在publisher服务的application.yml中添加配置：
spring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: admin # 用户名
    password: xxxx # 密码

2.2 在publisher中用RabbitTemplate实现消息发送
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, spring amqp!";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

3. 在consumer模块中搞事情
3.1首先配置MQ地址，在consumer服务的application.yml中添加配置
spring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: admin # 用户名
    password: xxxx # 密码

3.2然后在consumer服务的`cn.itcast.mq.listener`包中新建一个类SpringRabbitListener，代码如下：

package cn.itcast.mq.listener;

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("spring 消费者接收到消息：【" + msg + "】");
    }
}

测试：启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息4.2 Work Queue

Work queue，工作队列，可以提高消息处理速度，避免队列消息堆积。Work queues，也被称为（Task queues），任务模型。简单来说就是**让多个消费者绑定到一个队列，共同消费队列中的消息**。
Work模型的使用：

• 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
  
• 通过设置prefetch来控制消费者预取的消息数量
  

1.在publisher服务中的SpringAmqpTest类中添加一个测试方法：

/**
     * workQueue
     * 向队列中不停发送消息，模拟消息堆积。
     */
@Test
public void testWorkQueue() throws InterruptedException {
    // 队列名称
    String queueName = "simple.queue";
    // 消息
    String message = "hello, message_";
    for (int i = 0; i < 50; i++) {
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}

2.要模拟多个消费者绑定同一个队列
2.1 在spring中有一个简单的配置，可以解决这个问题(每个customer处理消息速度快慢不同的问题)。我
们修改consumer服务的application.yml文件，添加配置：
spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

2.2 我们在consumer服务的SpringRabbitListener中添加2个新的方法：

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}5. 发布订阅的模型（实现消息的群发）

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。
常见exchange类型包括：

• Fanout：广播-将消息交给所有绑定到交换机的队列
  
• Direct：定向-把消息交给符合指定routing key 的队列
  
• Topic：通配符-把消息交给符合routing pattern（路由模式） 的队列
  

5.1 Fanout交换机

• 创建一个交换机 itcast.fanout，类型是Fanout
  
• 创建两个队列fanout.queue1和fanout.queue2，绑定到交换机itcast.fanout
  

1.在publisher服务的SpringAmqpTest类中添加测试方法：

@Test
public void testSendDirectExchange() {
    // 队列名称
    String exchangeName = "itcast.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

2. 在consumer中创建一个类，声明队列和交换机
@Configuration
public class FanoutConfig {
    /**
     * 声明交换机
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

在consumer服务的SpringRabbitListener中添加两个方法，作为消费者：
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
    System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
    System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
}
5.2 Direct交换机

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下：

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个`RoutingKey`（路由key）
  
• 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 `RoutingKey`。
  
• Exchange不再把消息交给每一个绑定的队列，而是根据消息的`Routing Key`进行判断，只有队列的`Routingkey`与消息的 `Routing key`完全一致，才会接收到消息。
  

1.在publisher服务的SpringAmqpTest类中添加测试方法：

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "jonathan.direct";
    // 消息
    String message = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxx！";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "red", message); // red就是RoutingKey
}

2.基于注解声明队列和交换机
基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。
在consumer的SpringRabbitListener中添加两个消费者，同时基于注解来声明队列和交换机：

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue1"), // 声明队列
    exchange = @Exchange(name = "jonathan.direct", type = ExchangeTypes.DIRECT), // 声明交换机
    key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
    System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
}


@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue2"),
    exchange = @Exchange(name = "jonathan.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
}5.3 Topic交换机

`Topic`类型的`Exchange`与`Direct`相比，都是可以根据`RoutingKey`把消息路由到不同的队列。只不过`Topic`类型`Exchange`可以让队列在绑定`Routing key` 的时候使用通配符！
通配符规则：

• `#`：匹配一个或多个词 `item.spu.insert`
  
• `*`：匹配不多不少恰好1个词 `item.spu`
  

1. publisher中搞事情
@Test
public void testSendTopicExchange() {
    // 交换机名称
    String exchangeName = "jonathan.topic";
    // 消息
    String message = "xxxxxxxxxxxxxxxxxxx!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}


2.在consumer服务的SpringRabbitListener中添加方法：
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue1"),
    exchange = @Exchange(name = "jonathan.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"
))
public void listenTopicQueue1(String msg){
    System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue2"),
    exchange = @Exchange(name = "jonathan.topic", type = ExchangeTypes.TOPIC),
    key = "#.news"
))
public void listenTopicQueue2(String msg){
    System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
}6. 消息转换器(解决默认发送Map消息，不能转换，导致接受不到)

Spring会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。只不过，默认情况下Spring采用SimpleMessageConvert转换器的序列化方式是JDK序列化。JDK序列化存在下列问题：

• 数据体积过大
  
• 有安全漏洞
  
• 可读性差
  

显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。
1.在publisher和consumer两个服务中都引入依赖：

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId> // 实体类转为JSON数据
    <version>2.9.10</version>
</dependency>


2.配置消息转换器。在publisher和consumer启动类中添加一个Bean即可
@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter();
}