微服务（十一）——MQ

初识MQ

在学习MQ之前，我们先来总结下在业务中使用同步调用的优缺点。

优点：时效性较强，可以立即得到结果

缺点：

• 耦合度高：每次加入新的需求，都要修改原来的代码
• 性能下降：调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和
• 资源浪费：调用链中的每个服务在等待响应过程中，并不能释放请求占用的资源，高并发场景下会季度浪费系统资源
• 级联失败：如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障

如果是一些有性能要求、高并发的业务场景，就推荐使用异步通信调用了。

异步调用常见实现就是事件驱动模型。事件驱动模型引入了一个事件代理者Broker。

在事件驱动模型中，如果有人支付成功了，就认为是一个事件，支付服务会发布事件交给Broker去管理。其他服务按需订阅事件，当有新的事件时，Broker会通知订阅了此事件的服务进行各自的业务处理。在这个过程中，支付服务只管发布事件，而不用再去等待其它业务的执行。

总结下异步通信的优缺点

优点：

• 耦合度低
• 吞吐量提升
• 故障隔离
• 流量削峰

缺点：

• 依赖于Broker的可靠性、安全性、吞吐能力
• 架构复杂了，业务没有明显的流程线，不好追踪管理

MQ（MessageQueue），中文是消息队列，字面来看就是存放消息的队列，也就是我们事件驱动架构中的broker（用来存储事件，在这里我们称事件为消息。）

比较下目前常用的消息队列技术：

对于中小型企业，我们并不需要像Kafka那样非常高的吞吐量，我们一般更在意消息的可靠性和稳定性，我们本章主要学习了解下RabbitMQ。

RabbitMQ

安装入门

RabbbitMQ是基于Erlang语言开发的开源消息通信中间件，官网地址：https://www.rabbitmq.com

服务器上使用docker进行单机部署

拉取镜像

运行MQ容器（mq1：主机名，集群部署需要使用，这里可以不设置  15672：ui管理平台的端口  5672：消息通信的端口）

docker run \
 -e RABBITMQ_DEFAULT_USER=lsy \
 -e RABBITMQ_DEFAULT_PASS=123321 \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

查看运行结果，运行成功

打开浏览器访问

输入我们运行容器时配置的账号密码，登录

重点看下RabbitMQ中的几个概念

• channel：通道，操作MQ的工具
• exchange：交换机，路由消息到队列中
• queue：消息队列，用于缓存消息
• virtual host：虚拟主机，是对queue、exchange等资源的逻辑分组，可以理解成多租户的使用概念

发布者缓存消息到消息队列，消费者通过订阅消息获取消息队列的消息。

消息模型

MQ的官方文档中给出了5个MQ的Demo示例，对应了几种不同的用法

HelloWorld

官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色：

• publisher：消息发布者，将消息发送到队列queue
• queue：消息队列，负责接收并缓存消息
• consumer：订阅队列，处理队列中的消息

测试下

添加依赖

<!--rabbitmq 客户端-->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
</dependency>

发布消息类Send.java

package cn.lsy.mq;

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

import java.util.Scanner;

public class Send {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception  {
        ConnectionFactory factory = new ConnectionFactory();
        //配置rabbitmq连接工厂
        factory.setHost("111.229.191.140");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("lsy");
        factory.setPassword("xxxxxx");

        //创建连接
        Connection connection = factory.newConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            //消息内容
            String message = scanner.nextLine();
            //发布消息
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
        }

    }
}

订阅接收消息类Recv.java

package cn.lsy.mq;

import com.rabbitmq.client.*;

import java.nio.charset.StandardCharsets;

public class Recv {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("111.229.191.140");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("lsy");
        factory.setPassword("xxxxxx");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        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 '" + message + "'");
        };

        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息消费被中断" + consumerTag);
        };

        //true-自动应答  false-手动应答
        boolean autoAck = true;

        String consumerTag = channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, cancelCallback);
        System.out.println("注册到RabbitMQ中后，RabbitMQ给的唯一标识是：" + consumerTag);
    }
}

方法参数解释：

声明队列——AMQP.Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)

• queue：被绑定的消息队列名——当该消息队列不存在时，将新建该消息队列
• durable：是否持久化——是否将队列持久化到mnesia数据库中，有专门的表保存我们的队列声明。 该参数持久化的仅为队列，而不包含队列中的消息。
• exclusive：该通道是否独占该队列——①当前定义的队列是connection的channel是共享的，其他的connection是访问不到的。②当connection关闭的时候，队列将被删除。
• autoDelete：自动删除——消费完成时是否删除队列，该删除操作在最后一个consumer(消费者)彻底断开连接之后进行。
• arguments：其他配置参数——rabbitmq的一个扩展，功能强大

基础的发布消息——basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body)

• String exchange : 交换机名， 当不使用交换机时，传入“”空串。
• String routingKey :（路由地址） 发布消息的队列， 无论channel绑定那个队列，最终发布消息的队列都有该字串指定
• AMQP.BasicProperties props ：消息的配置属性，例如 MessageProperties.PERSISTENT_TEXT_PLAIN 表示消息持久化。
• byte[] body ：消息数据本体， 必须是byte数组

消息回调接口——DeliverCallback，RabbitMQ推送给消费者消息回调接口，在该接口中用于编写如何对消息进行处理。

• comsumerTag：消费者注册到RabbitMQ之后，RabbitMQ给生成的一个该消费者的唯一标识
• delivery：推送过来的消息的信息
•     -Properties：消息的属性信息
•     -Envelope：包装信息
•         -deliveryTag：消息的ID
•         -redeliver：是否重新投递,当RabbitMQ发现消费者无法应答的时候，RabbitMQ会将消息重新编排，进行重新投递
•         -exchange：所使用的交换机
•         -routingKey：路由KEY
•     -body：消息体

取消订阅回调——CancelCallback ，rabbitmq取消该消费者对信道中队列的订阅时或者我们在RabbitMQ管理界面手动删除该队列时，调用的回调接口。

消费者订阅队列——basicConsume，在某个队列上注册消费者，在这个队列中有消息时，就会把消息转发给此Channel来处理，如果这个队列有多个消费者，RabbitMQ则会采用轮转的方式将消息分发给各个消费者。

启动Send和Recv

同时在RabbitMQ管理控制台可以看到，队列hello自动创建

测试下消息发送和接收，可以正常接收到消息。

SpringAMQP

什么是SpringAMQP

首先，我们需要了解下什么是AMQP。

AMQP，全称Advanced Message Queuing Protocol，是用于在应用程序或之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。

而Spring AMQP，是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现。

特征：

• 侦听器容器，用于异步处理入站消息
• 用于发送和接收消息的RabbitTemplate
• RabbitAdmin用于自动声明队列，交换和绑定

Hello World

引入依赖

<!--amqp依赖 包含rabbitmq-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置连接信息application.yml

logging:
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS
spring:
  rabbitmq:
    host: 111.229.191.140 # rabbitMQ的ip地址
    port: 5672 # 端口
    username: lsy # 自己添加的用户名
    password: xxxxxx # 自己设置的密码
    virtual-host: / # 虚拟主机分组 没有新建这里就是默认的

测试类，加上RabbitListener注解是为了自动创建队列

package cn.lsy.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;


@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    private final String queueName = "simple.queue";


    @Test
    @RabbitListener(queuesToDeclare = @Queue(queueName))
    public void test() {
        String message = "hello,spring amqp!";
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

启动测试，进入rabbitmq管理端查看，成功创建了队列simple.queue，并且存入了一条消息

点进去查看，可以看到我们发布的消息数据

消费者接收消息

package cn.lsy.mq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SpringRabbitListener {

     @RabbitListener(queues = "simple.queue")
     public void listenSimpleQueue(String msg) {
         System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
     }
}

启动测试，成功接收到消息

注意：消息一旦消费就会从队列删除，RabbitMQ没有消息回溯功能

Work Queue

work queue，工作队列，多个消费者消费同一队列，可以提高消息处理速度，避免消息队列堆积。

代码其实和上面差不多，先发布n条消息到工作队列，然后创建多个消费者消费此队列

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    private final String queueName = "work.queue";


    @Test
    public void testWorkQueue() throws InterruptedException {
        String message = "hello,message__";
        for (int i = 1; i <= 50; i++) {
            rabbitTemplate.convertAndSend(queueName, message + i);
            Thread.sleep(20);
        }
    }
}

@Component
public class SpringRabbitListener {

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

    @RabbitListener(queues = "work.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }
}

prefetch，消费者预取的消息数，不配置默认为消息数/消费者数

这里要注意：

• 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
• 默认根据消费者数量平均分配消息，比如50条消息，两个消费者，则是每人25条。和消息处理速度无关。
• 为了更好的提高消息消费效率，可以通过设置prefetch来控制消费者预取的消息数量。（如上例，设置prefetch为1，则消费者每次只会预取一条消息，处理完消息再取一条，如此重复，这样处理消息快的消费者可以多拿多消费，提高消费效率）

发布订阅模型

发布订阅模式与之前案例的区别就是允许将统一消息发送给多个消费者。实现方式是加入了exchange（交换机）。加入交换机以后，消息会通过交换机转发到一个或多个队列，这个由交换机的类型所决定。

常见的交换机类型包括3种：

• Fanout：广播
• Direct：路由
• Topic：主题

交换机的作用

• 接收publisher发送的消息
• 将消息按照规则路由到与之绑定的队列
• 不能缓存消息，路由失败，消息丢失

Spring AMQP中提供了一些交换机相关的API

Fanout Exchange

Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue。

声明交换机、队列，并绑定

package cn.lsy.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {
    // 声明交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("lsy.fanout");
    }

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

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

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

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

启动项目，可以看到，交换机和队列已创建，并且成功绑定

消息发布

    @Test
    public void testSendFanoutExchange() {
        //交换机名称
        String exchangeName = "lsy.fanout";
        //消息内容
        String message = "hello,fanout msg!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName, "", message);
    }

消息订阅

package cn.lsy.mq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) {
        System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");
    }
    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) {
        System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】");
    }
}

启动测试，两个队列成功收到消息

Direct Exchange

Direct Exchange会将接收到的消息根据规则路由到指定的Queue，因此成为路由模式（routes）。

• 每一个Queue都与Exchange设置一个或多个BindingKey
• 发布者发送消息时，指定消息的RoutingKey
• Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
• 当存在BindingKey相同时，且指定消息的RoutingKey为此共有BindingKey时，效果其实和Fanout广播模式类似

利用@RabbitListener声明Exchange、Queue、RoutingKey（也可以使用Bean声明，不过要麻烦点）

package cn.lsy.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SpringRabbitListener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "lsy.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 = "lsy.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg) {
        System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
    }
}

启动项目，查看RabbitMQ管理后台，可以看到，声明和绑定成功

发布消息测试

    @Test
    public void testSendDirectExchange() {
        //交换机名称
        String exchangeName = "lsy.direct";
        //消息内容
        String message = "hello,blue!";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName, "blue", message);
    }

依次改成yellow、red测试

Topic Exchange

Topic Exchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以.分割，可以看成是DirectExchange的一种简化书写。

声明交换机、队列

package cn.lsy.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SpringRabbitListener {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "lsy.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 = "lsy.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
    }
}

启动查看，声明成功

发布消息测试

    @Test
    public void testSendTopicExchange() {
        //交换机名称
        String exchangeName = "lsy.topic";
        //消息内容
        String message = "中国有56个民族";
        //发送消息
        rabbitTemplate.convertAndSend(exchangeName, "china.nation", message);
    }

其它匹配可以自行测试

消息转换器

在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。

声明一个队列

@Configuration
public class FanoutConfig {

    @Bean
    public Queue objectQueue() {
        return new Queue("object.queue");
    }
}

启动项目，查看管理端

发送消息

    @Test
    public void testSendObjectQueue() {
        Map<String, Object> msg = new HashMap<>();
        msg.put("name", "张三");
        msg.put("age", "21");
        //发送消息
        rabbitTemplate.convertAndSend("object.queue", msg);
    }

测试，查看管理端，成功收到一条消息

查看消息详情，传输的是序列化后的对象信息，默认使用JDK自身的序列化

Spring对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。

如果要修改只需要定义一个MessageConverter类型的Bean即可。推荐使用JSON方式序列化。

引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

声明MessageConverter

@Bean
public MessageConverter messageConverter(){
    return new Jackson2JsonMessageConverter();
}

删除上一条消息，再发送测试，可以看到，成功转变成了json的序列化格式

接收消息

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String,Object> msg){
        System.out.println("接收到object.queue的消息：" + msg);
    }
}

启动测试，成功接收

注意：发送方和接收方在不同微服务里的话，必须使用相同的MessageConverter，否则会报错哦。