RabbitMQ

初始MQ

高性能异步通信组件

• 同步通讯：就如同打视频电话，双方的交互都是实时的。因此同一时刻你只能跟一个人打视频电话。
• 异步通讯：就如同发微信聊天，双方的交互不是实时的，你不需要立刻给对方回应。因此你可以多线操作，同时跟多人聊天。

两种方式各有优劣，打电话可以立即得到响应，但是你却不能跟多个人同时通话。发微信可以同时与多个人收发微信，但是往往响应会有延迟。

所以，如果我们的业务需要实时得到服务提供方的响应，则应该选择同步通讯（同步调用）。而如果我们追求更高的效率，并且不需要实时响应，则应该选择异步通讯（异步调用）。

同步调用：

拓展性差

性能下降

级联失败

异步调用：

异步调用方式其实就是基于消息通知的方式，一般包含三个角色：

• 消息发送者：投递消息的人，就是原来的调用方
• 消息Broker：管理、暂存、转发消息，你可以把它理解成微信服务器
• 消息接收者：接收和处理消息的人，就是原来的服务提供方

在异步调用中，发送者不再直接同步调用接收者的业务接口，而是发送一条消息投递给消息Broker。然后接收者根据自己的需求从消息Broker那里订阅消息。每当发送方发送消息后，接受者都能获取消息并处理。
这样，发送消息的人和接收消息的人就完全解耦了。

综上，异步调用的优势包括：

• 耦合度更低
• 性能更好
• 业务拓展性强
• 故障隔离，避免级联失败
• 缓存消息，流量削峰填谷

当然，异步通信也并非完美无缺，它存在下列缺点：

• 不能立即得到结果，时效性差

• 不确定下游业务是否成功

• 完全依赖于Broker（消息代理）的可靠性、安全性和性能

• 架构复杂，后期维护和调试麻烦

技术选型

消息Broker，目前常见的实现方案就是消息队列（MessageQueue），简称为MQ.
目比较常见的MQ实现：

• ActiveMQ
• RabbitMQ
• RocketMQ
• Kafka

几种常见MQ的对比：

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP，XMPP，SMTP，STOMP	OpenWire,STOMP，REST,XMPP,AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般（十万）	差	高	非常高（百万）
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

追求可用性：Kafka、 RocketMQ 、RabbitMQ
追求可靠性：RabbitMQ、RocketMQ
追求吞吐能力：RocketMQ、Kafka
追求消息低延迟：RabbitMQ、Kafka

据统计，目前国内消息队列使用最多的还是RabbitMQ，再加上其各方面都比较均衡，稳定性也好，因此我们课堂上选择RabbitMQ来学习。

RabbitMQ

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

安装

还去学了docker，傻瓜式安装

docker run \
 -e RABBITMQ_DEFAULT_USER=fangyuan\
 -e RABBITMQ_DEFAULT_PASS=123456\
 -v mq-plugins:/plugins \
 --name mq \
 --hostname mq \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3.8-management

可以看到在安装命令中有两个映射的端口：

• 15672：RabbitMQ提供的管理控制台的端口
• 5672：RabbitMQ的消息发送处理接口

RabbitMQ对应的架构图

其中包含几个概念：

• **publisher**：生产者，也就是发送消息的一方
• **consumer**：消费者，也就是消费消息的一方
• **queue**：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理
• **exchange**：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。
• **virtual host**：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue

上述这些东西都可以在RabbitMQ的管理控制台来管理

收发消息

我们打开Exchanges选项卡，可以看到已经存在很多交换机：

我们点击任意交换机，即可进入交换机详情页面。

队列queue

绑定关系

只有交换机绑定了相应的队列，才知道如何发送消息，给谁发送消息

数据隔离

不同的用户可以创建不同的虚拟主机，不同的虚拟主机之间是隔离的

SpringAMQP

AMQP ：Advanced Message Queuing Protocol

由于RabbitMQ采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。
但是，RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于SpringBoot对其实现了自动装配，使用起来非常方便。

快速入门

不过有时候为了测试方便，我们也可以直接向队列发送消息，跳过交换机。
在入门案例中，我们就演示这样的简单模型，如图：

也就是：

• publisher直接发送消息到队列
• 消费者监听并处理队列中的消息

:::warning
注意：这种模式一般测试使用，很少在生产中使用。
:::

配置MQ地址：

spring:
  rabbitmq:
    host: 192.168.239.129 # 你的虚拟机IP
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: fangyuan # 用户名
    password: 123456 # 密码

消息发送：

在测试类中实现

@SpringBootTest
public class RabbitMQ {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void test1(){
        String queueName = "hello.queue1";
        String mes = "hello,amqp!";
        rabbitTemplate.convertAndSend(queueName,mes);
    }
}

消息接收

@Slf4j
@Component
public class MqListener {
    // 利用RabbitListener来声明要监听的队列信息
    // 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
    // 可以看到方法体中接收的就是消息体的内容
    @RabbitListener(queues = "hello.queue1")
    public void listensSimpleQueue(String msg){
        System.out.println("消费者收到了hello.queue1的消息：" + msg);
    }
}

测试，接收成功

WorkQueues模型

Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。

当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。
此时就可以使用work 模型，多个消费者共同处理消息处理，消息处理的速度就能大大提高了。

接下来，我们就来模拟这样的场景。

要模拟多个消费者绑定同一个队列，我们在consumer服务的SpringRabbitListener中添加2个新的方法：

@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);
}

注意到这两消费者，都设置了Thead.sleep，模拟任务耗时：

• 消费者1 sleep了20毫秒，相当于每秒钟处理50个消息
• 消费者2 sleep了200毫秒，相当于每秒处理5个消息

测试结果：发现是轮循，一人处理一个

也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的

在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置：

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

再次测试，发现结果如下：

消费者1的处理结果是2的十倍了，符合我们设置的延时效果。提升了效率

交换机类型

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

交换机的类型有四种：

• Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
• Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
• Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
• Headers：头匹配，基于MQ的消息头匹配，用的较少。

Fanout交换机

Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。
在广播模式下，消息发送流程是这样的：

• 1）  可以有多个队列
• 2）  每个队列都要绑定到Exchange（交换机）
• 3）  生产者发送的消息，只能发送到交换机
• 4）  交换机把消息发送给绑定过的所有队列
• 5）  订阅队列的消费者都能拿到消息

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

@Test
public void testFanoutExchange() {
    // 交换机名称
    String exchangeName = "hmall.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

在consumer服务的SpringRabbitListener中添加两个方法，作为消费者：

@RabbitListener(queues = “fanout.queue1”)

这个注解里面不仅可以指定，队列的名称，也可以指定交换机的名称

@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 + "】");
}

Direct交换机

但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下：

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

rabbitTemplate.convertAndSend(exchangeName, “路由key”, message);

总结：

• Fanout交换机将消息路由给每一个与之绑定的队列
• Direct交换机根据RoutingKey判断路由给哪个队列
• 如果多个队列具有相同的RoutingKey，则与Fanout功能类似

Topic交换机

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

BindingKey 一般都是有一个或多个单词组成，多个单词之间以.分割，例如： item.insert

通配符规则：

• #：匹配一个或多个词
• *：匹配不多不少恰好1个词

举例：

• item.#：能够匹配item.spu.insert 或者 item.spu
• item.*：只能匹配item.spu

图示：

假如此时publisher发送的消息使用的RoutingKey共有四种：

• china.news 代表有中国的新闻消息；
• china.weather 代表中国的天气消息；
• japan.news 则代表日本新闻
• japan.weather 代表日本的天气消息；

解释：

• topic.queue1：绑定的是china.# ，凡是以 china.开头的routing key 都会被匹配到，包括：
  • china.news
  • china.weather
• topic.queue2：绑定的是#.news ，凡是以 .news结尾的 routing key 都会被匹配。包括:
  • china.news
  • japan.news

声明队列和交换机

在config中声明一个配置类，用@Configuration注解声明为配置类，@Bean表示在声明时候就加载在Bean对象中

SpringAMQP提供了一个Queue类，用来创建队列：

SpringAMQP还提供了一个Exchange接口，来表示所有不同类型的交换机：

/**
 * 声明交换机
 * @return Fanout类型交换机
 */
@Bean
public FanoutExchange fanoutExchange(){
    return new FanoutExchange("hmall.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);
   }

direct实例

direct模式由于要绑定多个KEY，会非常麻烦，每一个Key都要编写一个binding：

 /**
     * 声明交换机
     * @return Direct类型交换机
     */
    @Bean
    public DirectExchange directExchange(){
        return ExchangeBuilder.directExchange("hmall.direct").build();
    }
    /**
     * 第1个队列
     */
    @Bean
    public Queue directQueue1(){
        return new Queue("direct.queue1");
    }

/**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
    }

绑定交换机和队列的时候，key参数一次只能用with传递一次。太麻烦

基于注解声明

注解的嵌套，人嘛了，记不住再回来看

在注解@RabbitListener()中使用@Queuebinding（）声明队列，交换机，和绑定参数

导包别导错了，全是注解类型的

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

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

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue1"),
    exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"
))
public void listenTopicQueue1(String msg){
    System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue2"),
    exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
    key = "#.news"
))
public void listenTopicQueue2(String msg){
    System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
}

消息转换器

Spring的消息发送代码接收的消息体是一个Object：

而在数据传输时，它会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。
只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

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

配置JSON转换器

显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。

在publisher和consumer两个服务中都引入依赖：

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

注意，如果项目中引入了spring-boot-starter-web依赖，则无需再次引入Jackson依赖。

配置消息转换器，在publisher和consumer两个服务的启动类中添加一个Bean即可：

@Bean
public MessageConverter messageConverter(){
    // 1.定义消息转换器
    Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
    // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
    jackson2JsonMessageConverter.setCreateMessageIds(true);
    return jackson2JsonMessageConverter;
}

消息转换器中添加的messageId可以便于我们将来做幂等性判断。

对之前的Redis业务改造

把以前放入阻塞队列的消息，放入消息队列RabbitMQ中

引入依赖，注入RabbitTemplate对象

其中，convertAndSend的传入的参数为，交换机，关键字，和对象

    @Resource
    private RabbitTemplate rabbitTemplate;
/*.......*/

//放入消息队列RabbitMQ
        try {
            rabbitTemplate.convertAndSend("voucherOrder.topic","secKill.success",voucherOrder);
        } catch (AmqpException e) {
            log.info("消息发送失败，用户ID：{}",voucherOrder.getUserId());
        }
        //放进阻塞队列
        //orderTasks.add(voucherOrder);

新建一个Listener包，定义MQListener类，用于监听消息

注入之前的service层对象，直接调用其方法即可

@Slf4j
@Component
public class MqListener {

    @Resource
    private VoucherOrderServiceImpl voucherOrderService;

    // 利用RabbitListener来声明要监听的队列信息
    // 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
    // 可以看到方法体中接收的就是消息体的内容
    @RabbitListener(queues = "hello.queue1")
    public void listensSimpleQueue(String msg){
        System.out.println("消费者收到了hello.queue1的消息：" + msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "voucherOrder.queue"),
            exchange = @Exchange(name = "voucherOrder.topic",type = ExchangeTypes.TOPIC),
            key = "secKill.success"
    ))
    public void voucherOrderListener(VoucherOrder voucherOrder){
        //TODO 消费者拿到消息队列的消息，的处理方法
        //voucherOrderService.handleVoucherOrder(voucherOrder);
        voucherOrderService.createVoucherOrder(voucherOrder);
    }

}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "voucherOrder.queue"),
        exchange = @Exchange(name = "voucherOrder.topic",type = ExchangeTypes.TOPIC),
        key = "secKill.success"
))

使用注解嵌套注解的方式，来实现

具体解释如下：

1. @RabbitListener: 这是一个监听器注解，用于声明一个方法作为 RabbitMQ 的消息消费者。当 RabbitMQ 中有消息到达时，被注解的方法将被调用。
2. bindings: 这是 @RabbitListener 注解的一个属性，用于指定队列和交换机之间的绑定关系。在这里，通过 @QueueBinding 注解指定了队列与交换机之间的绑定关系。
3. @Queue: 在 @QueueBinding 注解中，@Queue 用于声明一个队列，并设置队列的属性。在这里，使用了 name 属性指定了队列的名称为 “voucherOrder.queue”。
4. @Exchange: 在 @QueueBinding 注解中，@Exchange 用于声明一个交换机，并设置交换机的属性。在这里，使用了 name 属性指定了交换机的名称为 “voucherOrder.topic”，type 属性指定了交换机的类型为 “topic”。
5. key: 在 @QueueBinding 注解中，key 属性用于指定路由键，表示消息从交换机发送到队列时所需匹配的路由规则。在这里，设置了 “secKill.success” 作为路由键。

综上所述，这段代码的作用是声明了一个 RabbitMQ 的消息监听器，监听名为 “voucherOrder.topic” 的交换机上的 “secKill.success” 路由键对应的队列 “voucherOrder.queue”，当有消息通过该路由键到达时，会触发被注解的方法进行消费。

来，上压力测试，好快啊，平均26ms

来三轮求取平均值 82ms

第三轮68ms

好像用了lua脚本后，就确保了一人一单不会被抢了，加Redisson分布式锁还有没有用了？

加锁，也是给后续写入数据库的操作加锁，感觉会没用，试试

果然，平均只有1s了