一、初识MQ

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

1.1常见的几种MQ

• ActiveMQ
• RabbitMQ
• RocketMQ
• Kafka

1.2几种MQ的对比

1.3MQ的基本结构

• 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.IOException;
import java.util.concurrent.TimeoutException;

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

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

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

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

        // 5.关闭通道和连接
        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 {
 
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

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

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

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                // 5.处理消息
                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什么是SpringAMQP

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

SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

3.2.1 Spirng AMQP提供的三个功能

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

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

在父工程中导入依赖坐标

        
            org.springframework.boot
            spring-boot-starter-amqp
        

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

spring:
  rabbitmq:
    host: 192.168.***.105
    port: 5672
    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");
    }

    //    声明队列1
    @Bean
    public Queue fanoutQueue1() {
 
        return new Queue("fanout.queue1");
    }

    //    声明队列2
    @Bean
    public Queue fanoutQueue2() {
 
        return new Queue("fanout.queue2");
    }

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

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

}

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

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

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

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

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

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

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

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

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

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

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

3.5 订阅/发布

模型结构如下：

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

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

3.6 Fanout Exchange的使用

模型：

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

• 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");
    }

    //    声明队列1
    @Bean
    public Queue fanoutQueue1() {
 
        return new Queue("fanout.queue1");
    }

    //    声明队列2
    @Bean
    public Queue fanoutQueue2() {
 
        return new Queue("fanout.queue2");
    }

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

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

}

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

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

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

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.queue1"),
            //exchange声明交换机，name交换机名字，type类型
            exchange = @Exchange(name = "chen.direct",type = "direct"),
            // 指定交换机向指定了bindingkey的队列发消息
            key = {
 "red","blue"}
    ))

3.7.2 演示

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

/**
     * @param msg
     * @return void
     * @author chenqingxu
     * @description 测试DirectExchange模式的交换机, 在@RabbitListener注解中声明对列以及交换机，如果Rabbit中没有，spring会帮我们创建
     * 注意，当消息中的routingkey为red时，两个队列都可收到消息
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name="direct.queue1"),
            exchange = @Exchange(name = "chen.direct",type = "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 = "chen.direct",type = "direct"),
            key = {
 "red","yellow"}
    ))
    public void listenDirectQueue2(String msg) {
 
        System.out.println("消费者收到direct.queue2的消息：" + 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[1]就是routingkey我是为了测试方便
        rabbitTemplate.convertAndSend(FanoutExchageName, split[1], 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演示

实现思路如下：

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

consumer中编写方法：

/**
     * @param msg
     * @return void
     * @author chenqingxu
     * @description 测试TopicExchange模式的交换机
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "chen.topic",type ="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 = "chen.topic",type ="topic"),
            key = {
 "#.news"}
    ))
    public void listenTopicQueue2(String msg) {
 
        System.out.println("消费者收到topic.queue2的消息：" + 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 msg = new HashMap<>();
    msg.put("name", "Jack");
    msg.put("age", 21);
    // 发送消息
    rabbitTemplate.convertAndSend("simple.queue","", msg);
}

结果：

总结：

JDK序列化存在下列问题：

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

解决方案

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

            com.fasterxml.jackson.core
            jackson-databind
        

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

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

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

package cn.itcast.mq;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
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 Jackson2JsonMessageConverter();
    }
}

2.消息接收端Consumer：

步骤一：导入依赖

步骤二：声明MessageConverter

前两步和上面的一样

步骤三：

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

需要有关Java资料或面试题可私！！！免费！！！