一、初识 MQ

MQ，中文意思是消息队列（MessageQueue），字面来看就是存放消息的队列，也就是事件驱动架构中的Broker。

1.1常见的几种MQ

ActiveMQ
rabbitMq
RocketMQ
Kafka

1.2几种MQ的对比

「Java基础」在SpringBoot环境，MQ消息队列&RabbitMQ的简单入门

1.3MQ的基本结构

「Java基础」在SpringBoot环境，MQ消息队列&RabbitMQ的简单入门

publisher ：消息的提供者
exchange：交换机，负责将消息传递给队列，不具备存储功能
queue：队列，用来暂时存放消息的，结构就是队列先进先出，消息取出后就在队列中消失，且不可回溯。
consumer：消费者，将消息从queue中取出
virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离

1.4.RabbitMQ的消息模型

基本消息队列（BasicQueue）
工作消息队列（WorkQueue）
发布订阅（Publisher、Subscribe），根据类型不同又分为三种：
Fanout Exchange：广播
Direct Exchange：路由
Topic Exchange：主题

2.小案列使用官方提供的方式

2.1Publisher实现

建立连接
创建Channel
声明队列
发送消息
关闭连接和channel
代码实现：

 package cn.itcast.mq.helloworld;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org. junit .Test;

import  java .io.IO Exception ;
import java.util.concurrent.TimeoutException;

public class PublisherTest {
 
    @Test
    public  void  testSendMessage() throws IOException, TimeoutException {
 
        //.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        //.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost(".168.150.101");
        factory.setPort();
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("");
        //.2.建立连接
        Connection connection = factory.newConnection();

        //.创建通道Channel
        Channel channel = connection.createChannel();

        //.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        //.发送消息
         String  message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.get Bytes ());
        System.out.println("发送消息成功：【" + message + "】");

        //.关闭通道和连接
        channel.close();
        connection.close();

    }
}

2.2 consumer 的实现

代码思路：

建立连接
创建Channel
声明队列
订阅消息

代码实现

 package cn.itcast.mq.helloworld;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerTest {
 

    public  static  void main(String[] args) throws IOException, TimeoutException {
 
        //.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        //.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost(".168.150.101");
        factory.setPort();
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("");
        //.2.建立连接
        Connection connection = factory.newConnection();

        //.创建通道Channel
        Channel channel = connection.createChannel();

        //.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        //.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                //.处理消息
                String message = new String(body);
                System.out.println("接收到消息：【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}

3.使用SpirngAMQP

文章中Consumer的监听类怎么使用，以为监听类注册成了配置类所以，我们只需运行Consumer这个服务即可。将接收队列消息启动，在去启动Publisher的测试类向队列中写数据即可，在consumer的控制台查看接收到的消息

3.1什么是AMQP

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

3.2什么是 spring AMQP

SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中 spring-amqp是基础抽象 ， spring-rabbit是底层的默认实现 。

Spring Amqp的官方地址：

3.2.1 Spirng AMQP提供的三个功能

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

3.3 使用Spring AMQP完成Basic Queue的实现

在父工程中导入依赖坐标

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

在YMAL配置文件配置连接RabbitMQ的信息

 spring:
  rabbitmq:
    host:.168.***.105
    port:
    virtual-host: / #虚拟主机，不要怀疑这台虚拟机就叫“/”
    username: chen
    password: ******

3.3.1编写Publisher的消息发送测试类：

 @Autowired
    RabbitTemplate rabbitTemplate;
    
 @Test
    public void testSpringAMQP() {
 
        String queryName = "queryChen";
        String message = "hello chen !!";
//        创建一个RabbitAdmin用于操作队列
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
//        声明一个队列
        rabbitAdmin.declareQueue(new Queue(queryName));
//       向队列里面发送消息
        rabbitTemplate.convertAndSend(queryName, message);
    }

3.3.2编写consumer的一个监听类

前置工作：

1.需要导入依赖，如果父工程有就不用导入了

2.编写配置文件和Publisher的配置文件内容一样

然后在consumer服务的 cn.itcast.mq.listener 包中新建一个类SpringRabbitListener，代码如下：

 @RabbitListener(queues = "queryChen")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException{
 
        System.out.println("收到消息："+msg);
    }

3.4使用Spring AMQP实现一个Work queues

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

3.4.1介绍一个声明消息队列的方法之一：配置类

使用配置类的方式让spring在创建时就将队列创建好，并将队列和交换机绑定好

以Fanout Exchange（后面会说）为例：

 package cn.itcast.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("chen.fanout");
    }

    //    声明队列
    @Bean
    public Queue fanoutQueue() {
 
        return new Queue("fanout.queue");
    }

    //    声明队列
    @Bean
    public Queue fanoutQueue() {
 
        return new Queue("fanout.queue");
    }

    //    声明绑定关系,绑定关系按照方法名称进行注入
//    绑定队列到交换机
    @Bean
    public Binding fanoutBinding(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
 
        return BindingBuilder
                .bind(fanoutQueue)
                .to(fanoutExchange);
    }

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

}

3.4.2在publisher的测试类中想队列添加多条消息

目的：50条数据，其中一个消费者处理消息速度快1秒钟处理50个，另个一慢，1秒钟处理10个，想让两个消费者共同完成这50个消息的处理。

 /**
     * @return void
     * @author chenqingxu
     * @description 演示发送条消息，让两个消费者获取
     */    @Test
    public void testSpringWorkAMQP() throws Exception {
 
        String queryName = "queryChen";
        String message = "hello chen__";
//        创建一个RabbitAdmin用于操作队列
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
//        声明一个队列
        rabbitAdmin.declareQueue(new Queue(queryName));
        for (int i =; i < 50; i++) {
 
//       向队列里面发送消息
            rabbitTemplate.convertAndSend(queryName, message + i);
            Thread.sleep();
        }
    }

3.4.3 在consumer中创建连个消费者与同一个队列绑定

 @RabbitListener(queues = "queryChen")
    public void listenWorkQueueMessage(String msg) throws InterruptedException {
 
        System.out.println("消费者收到消息：" + msg + "[" + LocalTime.now() + "]");
        Thread.sleep();
    }

    @RabbitListener(queues = "queryChen")
    public void listenWorkQueueMessage(String msg) throws InterruptedException {
 
        System.err.println("消费者...........收到消息：" + msg + "[" + LocalTime.now() + "]");
        Thread.sleep();
    }

知识点补充：此处并没有像我们想的那样，快的消费者处理的比慢的消费者多，而是两个消费者平均处理了队列中的消息。

原因：因为有一个消息预取机制，两个消费者在处理消息的同时，队列会按默认的轮询策略一直给消费者发送，预取机制没有上限，如果只有两个消费者这样就形成了平分队列中消息的情况。

解决：让处理快的消费者多处里就需要限制他们的预拉取， 在consumer的 配置文件中设置：

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

这样处理，速度快的消费者就可以处理的多。

3.5 订阅/发布

模型结构如下：

「Java基础」在SpringBoot环境，MQ消息队列&RabbitMQ的简单入门

消息的发送过程发生了改变：

Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：Fanout：广播，将消息交给所有绑定到交换机的队列Direct：定向，把消息交给符合指定routing key 的队列Topic：通配符，把消息交给符合routing pattern（路由模式）的队列
Consumer：消费者，与以前一样，订阅队列，没有变化
Queue：消息队列也与以前一样，接收消息、缓存消息。
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失

3.6 Fanout Exchange的使用

模型：

「Java基础」在SpringBoot环境，MQ消息队列&RabbitMQ的简单入门

在广播模式下，消息发送流程是这样的：

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

3.6.1演示

创建一个交换机chen.fanout，类型是Fanout
创建两个队列fanout.queue1，fanout.queue2 绑定到交换机chen.fanout
1.在consumer类中声明交换和队列并进行绑定
在springboot启动时就会自动在rabbitMq中创建队列和交换机并完成绑定

 package cn.itcast.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("chen.fanout");
    }

    //    声明队列
    @Bean
    public Queue fanoutQueue() {
 
        return new Queue("fanout.queue");
    }

    //    声明队列
    @Bean
    public Queue fanoutQueue() {
 
        return new Queue("fanout.queue");
    }

    //    声明绑定关系,绑定关系按照方法名称和属性进行依赖注入
//    绑定队列到交换机
    @Bean
    public Binding fanoutBinding(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
 
        return BindingBuilder
                .bind(fanoutQueue)
                .to(fanoutExchange);
    }

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

}

2.Fanout Exchange的使用，在consumer中编写接收消息功能

 /**
     * @param msg
     * @return void
     * @author chenqingxu
     * @description 队列交换机的使用，消费者处理速度快，消费者2处理速度慢
     */    @RabbitListener(queues = "fanout.queue")
    public void listenFanoutQueueMessage(String msg) throws InterruptedException {
 
        System.err.println("消费者...........收到消息：" + msg + "[" + LocalTime.now() + "]");
        Thread.sleep();
    }

    @RabbitListener(queues = "fanout.queue")
    public void listenFanoutQueueMessage(String msg) throws InterruptedException {
 
        System.err.println("消费者...........收到消息：" + msg + "[" + LocalTime.now() + "]");
        Thread.sleep();
    }

3.Fanout Exchange的使用，在publisher测试类中中编写消息发送功能

 /**
     * @return void
     * @author chenqingxu
     * @description 将消息发送给交换机，有交换机将消息放入队列，不在需要指定队列的名字
     */    @Test
    public void testSentFanoutExchange() {
 
//        准备交换机名称
        String FanoutExchageName = "chen.fanout";
//        需要发送的消息
        String message = "hello everyone !";
//        发送消息
        rabbitTemplate.convertAndSend(FanoutExchageName, "", message);
    }

总结：

交换机的作用是什么？

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

声明队列、交换机、绑定关系的Bean是什么？

Queue
FanoutExchange
Binding

3.7 Direct Exchange的使用

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

模型：

在Direct模型下：

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

3.7.1创建交换机和队列并将二者绑定的方式二：注解

在消费者端Consumer中设置的

 @RabbitListener(bindings = @QueueBinding(
//          value声明队列
            value = @Queue(name="direct.queue"),
            //exchange声明交换机，name交换机名字，type类型
            exchange = @Exchange(name = "chen.direct",type = "direct"),
            // 指定交换机向指定了bindingkey的队列发消息
            key = {
 "red","blue"}
    ))

3.7.2 演示

利用@RabbitListener声明Exchange、Queue、RoutingKey
在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
在publisher中编写测试方法，向chen. direct发送消息
Consumer服务：

 /**
     * @param msg
     * @return void
     * @author chenqingxu
     * @description 测试DirectExchange模式的交换机, 在@RabbitListener注解中声明对列以及交换机，如果Rabbit中没有，spring会帮我们创建
     * 注意，当消息中的routingkey为red时，两个队列都可收到消息
     */    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name="direct.queue"),
            exchange = @Exchange(name = "chen.direct",type = "direct"),
            key = {
 "red","blue"}
    ))
    public void listenDirectQueue(String msg) {
 
        System.out.println("消费者收到direct.queue的消息：" + msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name="direct.queue"),
            exchange = @Exchange(name = "chen.direct",type = "direct"),
            key = {
 "red","yellow"}
    ))
    public void listenDirectQueue(String msg) {
 
        System.out.println("消费者收到direct.queue的消息：" + msg);
    }

Publisher在测试类中实现发送消息功能

 /**
     * @return void
     * @author chenqingxu
     * @description 将消息发送给direct交换机，指定绑定的key
     */    @Test
    public void testSentDirectExchange() {
 
//        准备交换机名称
        String FanoutExchageName = "chen.direct";
//        需要发送的消息
//         String message = "hello,blue";
//        String message = "hello,yellow";
        String message = "hello,red";
//        发送消息,routingKey就是direct中和交换机绑定的队列的key
        String[] split = message.split(",");
        //split[]就是routingkey我是为了测试方便
        rabbitTemplate.convertAndSend(FanoutExchageName, split[], message);
    }

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

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

基于@RabbitListener注解声明队列和交换机有哪些常见注解？

@Queue
@Exchange

3.8Topic Exchange的使用

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

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

通配符规则：

# ：匹配一个或多个词或0个

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

举例：

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

item.* ：只能匹配 item.spu

模型：

Queue1：绑定的是 china.# ，因此凡是以 china. 开头的 routing key 都会被匹配到。包括china.news和china.weather
Queue2：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配。包括china.news和japan.news

3.8.1演示

实现思路如下：

并利用@RabbitListener声明Exchange、Queue、RoutingKey
在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
在publisher中编写测试方法，向chen. topic发送消息

consumer中编写方法：

 /**
     * @param msg
     * @return void
     * @author chenqingxu
     * @description 测试TopicExchange模式的交换机
     */    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue"),
            exchange = @Exchange(name = "chen.topic",type ="topic"),
            key = {
 "china.#"}
    ))
    public void listenTopicQueue(String msg) {
 
        System.out.println("消费者收到topic.queue的消息：" + msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue"),
            exchange = @Exchange(name = "chen.topic",type ="topic"),
            key = {
 "#.news"}
    ))
    public void listenTopicQueue(String msg) {
 
        System.out.println("消费者收到topic.queue的消息：" + msg);
    }

Publisher的测试类中编写测试方法

 /**
     * @return void
     * @author chenqingxu
     * @description 将消息发送给topic交换机，指定绑定的rountingkey
     */    @Test
    public void testSentTopicExchange() {
 
//        准备交换机名称
        String FanoutExchageName = "chen.topic";
//        需要发送的消息
//         String message = "hello,blue";
//        String message = "hello,yellow";
        String message = "China is awesome";
//        发送消息,routingKey就是topic中和交换机绑定的队列的key
        rabbitTemplate.convertAndSend(FanoutExchageName, "china.new", message);
       // rabbitTemplate.convertAndSend(FanoutExchageName, "china.news", message);
    }

3.9 Spirng-AMQP消息转换器

Spring的消息对象的处理是由 org.springframework.amqp.support.converter.MessageConverter 来处理额。而默认实现是 SimpleMessageConverter ，基于JDK的 ObjectOutStream 完成序列化。如果要修改，需要重新定义一个 MessageConverter 类型的Bean即可。推荐用JSON方式序列化。

测试默认转换器：

消息发送：

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

结果：

总结：

JDK序列化存在下列问题：

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

解决方案

使用JSON序列化，我们可以在父工程导入，也可以在两个子工程中都导入：

 <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

1.在消息发送方Publisher 编写配置类，也可不写配置类，写在启动类中 ：

步骤一：先导入依赖，可以导入父工程中，也可以在提供者的工程中导入，声明在父工程最好，因为消息接收服务也需要导入依赖

步骤二：在publisher服务中声明MessageConverter，这样就可以覆盖掉默认的消息转换器，这是springBoot的特性

 package cn.itcast.mq;

import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class PublisherApplication {
 
    public static void main(String[] args) {
 
        SpringApplication.run(PublisherApplication.class);
    }
    @Bean
    public MessageConverter jsonMessageConverter(){
 
              return new JacksonJsonMessageConverter();
    }
}

2.消息接收端Consumer：

步骤一：导入依赖

步骤二：声明MessageConverter

前两步和上面的一样

步骤三：

接收消息的类型，发送方发的什么类型的消息，我们就用什么类型来接收