消息队列

1.作用

在大多应用中，通过消息服务中间件来提升系统异步通信，扩展解耦能力。
消息服务（消息代理和目的地）
- 当消息发送者发送消息以后，将由消息代理管理，然后消息代理把消息传递到目的地
消息队列的俩种目的地
- 1.队列（queue）：点对点消息通信（p-to-p）
- 2.主题（topic）：发布/订阅消息通信（p/s）

2.消息的消费方式

1.点对点式
- 生产者生产并发送消息，消息代理将这些产生的消息放入队列中，消费者从队列这把这些消息获取并消费掉，当消息被读取后将被移出队列
- 消息只有唯一的发送者和接受者，但并不是只能有一个接收者。简记：一生一代多消费
2.发布订阅式
- 发送者(发布者)发送消息到主题，多个接收者(订阅者)监听(订阅)这个主题，那么接收者就会在消息到达时间同时收到消息。类似于微信公众号，微信用户订阅这个公众号(主题)，当公众号拥有者将消息发送到公众号，关注了这个公众号的所有人都可以同时收到消息
3.JMS(Java Message Service) JAVA消息服务
- 基于JVM消息代理的规范。ActiveMQ，HornetMQ是JMS实现
4.AMQP(Advanced Message Queuing Protocol)
- 高级消息队列协议，也是一个消息代理的规范，兼容JMS
- RabbitMQ就是AMQP的实现

3.支持

1.Spring支持
- spring-jms提供了对JMS的支持
- spring-rabbit提供了对AMQP的支持
- 通过实现ConnectionFactory工厂来连接消息代理
- 提供JmsTemplate，RabbitTemplate来发送消息
- @JmsListener(JMS)，@RabbitListener（AMQP）注解在方法上监听消息代理发布的消息
- @EnableJms，@EnableRabbit开启支持
2.SpringBoot自动配置
- JmsAutoConfiguration
- RabbitAutoConfiguration

4.RabbitMQ使用及概念

1.十大核心

1.Message（消息）：由消息头和消息体组成。消息体是不透明的，消息头由routing-key(路由键)，priority(相对于其他消息的优先权)，delivery-mode(指出该消息可能需要持久化存储)等可选属性组成
2.Publisher：消息的生产者，也是一个向交换器发布消息的客户端应用程序。
3.Exchange：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。Exchange有四种类型：direct(默认)，fanout，topic和headers，类型不同，转发消息的策略也不同
4.Queue：消息队列，用来保存消息直到发送给消费者。一个消息可放入一个或多个队列中，消息如果没有被消费，将一直存在队列里面。
5.Binding：绑定，用于消息队列和交换器之间的关联。所有交换器可以理解成一个由绑定构成的路由表。Exchange和Queue的绑定可以是多对多的关系。
6.Connection：网络连接，比如一个TCP连接。
7.Channel：信道，多路复用连接中的一条独立的双向数据流通道。
8.Consumer：消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
9.Virtual Host：虚拟主机，表示一批交换器，消息队列和相关对象。每个vhost本质上是一个小型的RabbitMQ服务器，vhost是AMQP概念的基础，必须在连接时指定，RabbitMQ默认的vhost是/。
10.Broker：表示消息队服务器实体。

注：在使用RabbitMQ图形管理时（程序控制也一样），交换器中的routing key中的路由规则#代表匹配一个或多个单词，*代表匹配一个单词

2.springboot整合并使用RabbitMQ

1.导入所需的pom依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>

2.简单测试使用

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private AmqpAdmin amqpAdmin;

    @Test
    public void create(){
        // 创建交换器
        amqpAdmin.declareExchange(new DirectExchange("amqpadmin.exchange"));
        // 创建队列
        amqpAdmin.declareQueue(new Queue("amqpadmin.queue", true));
        // 创建绑定规则
        amqpAdmin.declareBinding(new Binding("amqpadmin.queue", Binding.DestinationType.QUEUE, "amqpadmin.exchange", "amqp.hhh", null));
        // 删除队列
        amqpAdmin.deleteQueue("amqpadmin.queue");
        // 删除交换器
        amqpAdmin.deleteExchange("amqpadmin.exchange");

    }

     /**
     * 1.单播（点对点）
     */
    @Test
    public void sendMessagePTP() {
        // Message需要自己构造一个,定义消息
        // rabbitTemplate.send(exchange, routeKey, message);

        // object默认当成消息体，只需要传入要发送的对象，自动序列化发送给rabbit
        // rabbitTemplate.convertAndSend(exchange, routeKey, object);
        Map<String, Object> map = new HashMap<>();
        map.put("msg", "小小侦探团");
        map.put("data", Arrays.asList("柯南", "灰原哀", 123, true));
        // 对象被默认序列化以后发送出去
        rabbitTemplate.convertAndSend("exchange.direct", "moying.news", map);
    }

    // 接受数据，如果想要json数据，在配置里面配置
    @Test
    public void receive(){
        Object o = rabbitTemplate.receiveAndConvert("moying.news");
        System.out.println(o.getClass());
        System.out.println(o);
    }

    /**
     * 广播发送消息
     */
    @Test
    public void sendMsg(){
        rabbitTemplate.convertAndSend("exchange.fanout", "", new Book("奥特曼", "lily"));
    }

配置json序列化的config如下：

package com.moying.config;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MyAMQPConfig {

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

注：在使用springboot测试类测试的时候，注意springboot的版本，如果版本是2.2以下，是要导入junit5，并且在使用单元测试的时候可能会发生自动注入为空的情况，导致空指针异常，因此我使用的是2.2以上版本。

3.springboot使用注解消费消息
在启动类开启RabbitMQ注解
@EnableRabbit // 开启基于注解的RabbitMQ模式
在方法上使用注解监听

    // 只有队列中有注解就调用这个方法
    @RabbitListener(queues = "moying.news")
    public void receive(Book book){
        System.out.println("收到消息：" + book);
    }

    // 监听消息的body和消息的信息
    @RabbitListener(queues = "moying")
    public void receiveMsg(Message message){
        System.out.println(message.getBody());
        System.out.println(message.getMessageProperties());
    }