RabbitMQ

异步通信-RabbitMQ

1、面临问题

    之前的请求都是同步调用虽然时效性较强，可以立即得到结果
    但是同步调用存在以下问题：
    1.耦合度高
    2.性能和吞吐能力下降
    3.有额外的资源消耗
    4.有级联失败问题

2、解决方案-RabbitMQ

2.1、简介

    MQ全称为Message Queue，即消息队列。“消息队列”是在消息的传输过程中保存消息的容器。它是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。
    官网：https://www.rabbitmq.com/
    参考：https://blog.csdn.net/kavito/article/details/91403659

2.2、主流产品比较

2.3、优缺点

【优点】

    1
    3.调用间没有阻塞，不会造成无效的资源占用
    4.耦合度极低，每个服务都可以灵活插拔，可替换
    5.流量削峰：不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件

【缺点】

    1.架构复杂了，业务没有明显的流程线，不好管理
    2.需要依赖于Broker(MQ)的可靠、安全、性能

2.4、运行原理

2.5、运行流程

【生产者发送消息流程】

1、生产者和Broker建立TCP连接。
​
2、生产者和Broker建立通道。
​
3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。
​
4、Exchange将消息转发到指定的Queue（队列）

【消费者接收消息流程】

1、消费者和Broker建立TCP连接
​
2、消费者和Broker建立通道
​
3、消费者监听指定的Queue（队列）
​
4、当有消息到达Queue时Broker默认将消息推送给消费者。
​
5、消费者接收到消息。
​
6、ack回复

3、使用步骤

3.1、导入依赖

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

3.2、配置

spring:
      rabbitmq:
        host: 192.168.248.222 # 主机名
        port: 5672 # 端口
        virtual-host: / # 虚拟主机
        username: itcast # 用户名
        password: 123321 # 密码

注意：RabbitMQ是可以支持多个交换机绑定一个队列的，也就是说比如test1交换机可以绑定通道aa，然后test2可是可以绑定通道aa，但是如果通道aa在第一次链接被建立的是Topic类型的，那么交换机想要链接也要是Topic类型的，如下图

3.3、BasicQueue 简单队列模型

生产者对应一个队列，这样的话就不需要我们设置交换机，会有一个默认的交换机

    生产者----消费者

【publisher】

   @Autowired
    private RabbitTemplate rabbitTemplate;  
​
    @Test
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, spring amqp!";
        // 发送消息
       //  第一个参数：想要去存储数据到的队列名称
       //  第一个参数：存储到队列中的消息
        rabbitTemplate.convertAndSend(queueName, message);
    }

【consumer】

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

3.4、WorkQueue 工作模式

    生产者----消费者1|消费者2

【publisher】

 /**
         * 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);
        }
    }

【consumer】

 /**
     * 这个的消费者中和下边消费者消费的是同一个队列，那么他们就是工作模式，因为工作模式就是：一个链接，一个生产者多个消费者
     * @param msg
     * @throws InterruptedException
     * 睡眠是为了实现能者多老，这样第一个就是慢的那个，第二个就是快的那个，快的那个就应该能者多劳
     */
    @RabbitListener(queues = "fanout.queue251")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("第一个的："+msg);
//        睡眠是为了实现能者多老，这样第一个就是慢的那个，第二个就是快的那个，快的那个就应该能者多劳
        Thread.sleep(200);
    }

    @RabbitListener(queues = "fanout.queue252")
    public void listenSimpleQueueMessage1(String msg) {
        System.out.println("第二个的："+msg);
        try {
//            睡眠是为了实现能者多老，这样第一个就是慢的那个，第二个就是快的那个，快的那个就应该能者多劳
            Thread.sleep(20);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

这样的话我们只需要设置两个消费者都去指向这个管道就行，其实这样是不需要交换机的，因为就一个管道，有一个默认的交换机

但是工作模式他有一个机制是“预取”机制，就是我们的消息弄到队列中去之后，消费者消费的时候，他是优先把队列的信息分配好之后，然后消费者再去消费，这样是不好的，因为比如上图的consumer2是效率不太行的，那么他执行的量和consumer1是一样的，这样不如按劳分配，能干的多干，不能干的少干，所以我们可以配置一个东西，让他们每次都只拿一个，拿完一个消费完了再去通道中拿。

【yml】

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

3.5、发布/订阅模式

① Fanout-扇出/广播模式

先说fanout模式，也叫广播模式，生产者给交换机信息，交换机会将信息分为两份分别这两个管道，这样两个管道拿到的信息都是一样的：
首先他是一个交换机对应两个管道，这是就需要我们创建交换机，然后将创建好的管道绑定到交换机中昂，那么我们在创建交换机的时候就可以绑定我们的管道，这样生产者在生产信息给交换机，交换机就会给管道分配信息

fanout代码实现，有两种方式，一种是在config中配置，一种是在注解中写：

@Configuration
public class FanoutConfig {
   
    // 声明交换机 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);
    }
​
    // 第1个队列
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
​
    // 绑定队列和交换机
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

【publisher】

  // fanout
    @Test
    public void testFanoutExchange() {
        // 队列名称
        String exchangeName = "itcast.fanout";
            // 消息
           //  第一个参数：想要去存储数据到的队列名称
           //  第一个参数：存储到队列中的消息
        String message = "hello, everyone!";
        rabbitTemplate.convertAndSend(exchangeName, "", message);
    }

【consumer】

  // fanout
    @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。

【publisher】

  // direct
    @Test
    public void testSendDirectExchange() {
        // 交换机名称
        String exchangeName = "itcast.direct";
        // 消息
        String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
        // 发送消息
             /**
             * 参数一: 交换机名称
             * 参数二: 路由信息
             * 参数三: 消息内容
             */
        rabbitTemplate.convertAndSend(exchangeName, "red", message);
    }

【consumer】

/**
 * 使用注解来创建交换机和队列和指定路由
 *value = @Queue(name = "direct.queue1"),    队列名称
 *exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),     交换机和交换机的类型
 *key = {"red", "blue"}   这个队列规定key也就是路由
 * @param msg
 */

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

③ Topic-主题模式

Topic

类型的

Exchange

与

Direct

相比，都是可以根据

RoutingKey

把消息路由到不同的队列。只不过

Topic

类型

Exchange

可以让队列在绑定

Routing key

的时候使用通配符！

Routingkey

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

item.insert

通配符规则：

#

：匹配一个或多个词

*

：匹配不多不少恰好1个词

【publisher】

// topic
    @Test
    public void testSendTopicExchange() {
        // 交换机名称
        String exchangeName = "itcast.topic";
        // 消息
        String message = "喜报！孙悟空大战哥斯拉，胜!";
        // 发送消息
            /**
             * 参数一: 交换机名称
             * 参数二: 路由信息   因为这里是传的有news后缀，所以在下边消费者中会路由到 
             * listenTopicQueue2方法中
             * 参数三: 消息内容
             */
        rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
    }  

【consumer】

/**
 * 使用注解来创建交换机和队列和指定路由
 *value = @Queue(name = "topic.queue2"),   队列名称
 *exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),     
 *交换机和交换机的类型
 *key = "#.news"   这个队列规定key也就是路由,这里意思就是如果后缀满足 .news 为结         *尾，那么就会路由到这个队列中
 * @param msg
 */

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

3.6、消息转换器

① 消息是对象

    问题:当消息传递是对象时，会调用jdk的序列化，内存空间占用大

【publisher】

// 测试jdk序列化
    @Test
    public void testSendMap() throws InterruptedException {
        // 准备消息
        Map<String,Object> msg = new HashMap<>();
        msg.put("name", "Jack");
        msg.put("age", 21);
        // 发送消息
        // messageConverter.toMessage(msg, msg);
        rabbitTemplate.convertAndSend("simple.queue","", msg);
    }

【consumer】

  // 测试jdk序列化
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(Map msg) throws InterruptedException {
        System.out.println("spring 消费者接收到消息：【" + msg + "】");
    }

② 解决

 <!--json转换-->
    <dependency>
        <groupId>com.fasterxml.jackson.dataformat</groupId>
        <artifactId>jackson-dataformat-xml</artifactId>
        <version>2.9.10</version>
    </dependency>   

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