同步与异步


解读：

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

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

  1. 同步常用于两个业务之间有联系，例如执行业务B 需要执行业务A后的结果 这种的都需要使用同步
  2. 异步用于 几个业务之间并没有直接联系，这时候异步更好的提供性能
    例如这个服务，更新交易流水肯定需要在扣减余额成功后执行，所以交易流水和扣减余额肯定要同步执行
    交易服务，通知服务，积分服务等之间并没有必要的联系，这些服务都可以异步执行

RabbitMQ的安装

  • 使用docker快速安装
    在docker中执行这段命令，可能会出容器启动失败的情况，那就下载最新版即可解决
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    docker run \
     -e RABBITMQ_DEFAULT_USER=itheima \
     -e RABBITMQ_DEFAULT_PASS=123321 \
     -v mq-plugins:/plugins \
     --name mq \
     --hostname mq \
     -p 15672:15672 \
     -p 5672:5672 \
     --network hm-net\
     -d \
     rabbitmq:3.8-management
  • 访问15672端口
  1. 可能会出现访问不了情况，这是因为没有开启插件
    执行这两行代码即可
    1
    2
    docker exec -it 容器名 bash
    rabbitmq-plugins enable rabbitmq_management
  2. 可能会出现身份验证的情况

    这是因为在创建容器的时候，添加用户的命令没有成功，需要重新添加用户
    1
    2
    rabbitmqctl add_user 用户名 密码
    rabbitmqctl  set_user_tags 用户名 administrator

安装成功

MQ的简单了解

MQ消息队列的架构
其中包含几个概念：

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

SpringAMQP

SpringAMQP 提供了java编程的方式操作RabbitMQ

SpringAMQP提供了三个功能：

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

    快速入门

  1. 导入 AMQP的依赖
    1
    2
    3
    4
    5
    <!--AMQP依赖，包含RabbitMQ-->
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

  2. 添加配置

    1
    2
    3
    4
    5
    6
    7
    spring:
      rabbitmq:
        host: 192.168.217.128 # 你的虚拟机IP
        port: 5672 # 端口
        virtual-host: / # 虚拟主机
        username: lhg # 用户名
        password: 123456 # 密码

  3. 简单演示
    演示过程中为了方便会省略转发器的部分，由纤细发送者直接发送到队列中

  • 客服端创建一个队列
  • 消息发送
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    package com.itheima.publisher.amqp;

    import org.junit.jupiter.api.Test;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;

    @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);
        }
    }
  • 消息接收
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    package com.itheima.consumer.listener;

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

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

Work Queues 模型

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

Work Queues 经常用于消息很多，消息可能堆积的形况使用，因为多个消费者可以更快的处理消息

Work Queues 消息队列中默认消息是平均分配的


例如由两个消费者A,B 消息队列中有50条消息，那么消费者A接收的消息可能是第1，3，5等消息，B接收到的是第2，4，6消息，这50条消息会轮询平均分配给消费者。
但是在实际开发中会有多实例的消费者，每个消费者的电脑性能不一样，所以处理消息的速度不一样，如果仍使用平均分配会不太合理

能者多劳

RabbitMQ默认情况会轮询平均分配，简单来说也就是所有消费者排好队，到队列中领取消息，每个消费者每次只能领一条消息，每次领完之后就到消费者队伍的末尾继续排队。
RabbitMQ也为我们提供了对应的 prefetch配置, prefetch配置表示每个消费者最多领取到的消息，设置为1就表示消费者领取一个消息后必须处理完这个消息才能再去排队，这就实现了能者多劳

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

交换机

交换机可以接收消息发送者发过来的消息，然后交换机可以进行一些处理，再转发到消息队列中

可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：

  • Publisher：生产者，不再发送消息到队列中，而是发给交换机
  • Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
  • Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
  • Consumer：消费者，与以前一样，订阅队列，没有变化

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

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

    Fanout交换机

    Fanout模式的交换机可以实现一个消息多次处理,这是Fanout交换机的特点，Fanout交换机将接收到的消息发送个所有的队列，不同的队列有相同的消息，这样就可以实现消息的多次处理了

    一个消息队列中的消息只能被消费者处理一次

  1. 创建交换机

    遇见的问题，点击交换机会报错状态码500

    解决：

    1. 进入容器 docker exec -it rabbit（容器ID 或者 名称） bash
    2. 进入配置目录 cd /etc/rabbitmq/conf.d/ 执行命令
      echo management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf
    3. 退出容器 exit
    4. 重启服务 docker restart rabbit（容器ID 或者 名称）

  1. 交换机绑定消息队列
  2. 消息发送
    1
    2
    3
    4
    5
    6
    7
    8
    @Test
    public void testFanoutExchange() {
        // 交换机名称
        String exchangeName = "hmall.fanout";
        // 消息
        String message = "hello, everyone!";
        rabbitTemplate.convertAndSend(exchangeName, "", message);
    }
  3. 消息接收
    1
    2
    3
    4
    5
    6
    7
    8
    9
    @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 交换机

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

Direct 交换机再绑定消息队列时候需要多指定一个Key,消息发送者发送消息到交换机的时候也需要多一个Key的参数，Direct交换机会比较这两个Key，将消息转发Key相同的消息队列中


在Direct模型下：

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
  • 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
  • Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息
    总结：
    描述下Direct交换机与Fanout交换机的差异？
  • 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
      总结

描述下Direct交换机与Topic交换机的差异？

  • Topic交换机接收的消息RoutingKey必须是多个单词，以 . 分割
  • Topic交换机与队列绑定时的bindingKey可以指定通配符

  • ：代表0个或多个词

  • *：代表1个词

java声明交换机和队列

在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时，队列和交换机是程序员定义的，将来项目上线，又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来，交给运维。在这个过程中是很容易出现错误的。
因此推荐的做法是由程序启动时检查队列和交换机是否存在，如果不存在自动创建

bean声明

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

我们可以自己创建队列和交换机，不过SpringAMQP还提供了ExchangeBuilder来简化这个过程：

而在绑定队列和交换机时，则需要使用BindingBuilder来创建Binding对象：

注解声明

实例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@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 + "】");
}