消息队列概述

---

消息队列MQ

---

• MQ全称为Message Queue消息队列是应用程序之间的通信方法
• 为什么使用MQ
  • 在项目中可将一些无需即时返回且耗时的操作提取出来进行异步处理而这种异步处理的方式大大的节省了服务器的请求响应时间从而提高了系统的吞吐量
• 开发中消息队列通常有如下应用场景
  • 1、任务异步处理
    • 将不需要同步处理的且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间
  • 2、应用程序解耦合
    • MQ相当于一个中介生产方通过MQ与消费方交互它将应用程序进行解耦合

AMQP和JMS

• MQ 是消息通信的模型实现MQ大致有两种主流方式AMQP、JMS
• AMQP
  • AMQP是一种协议更准确的说是一种binary wire-level protocol链接协议这是其与JMS的本质差别AMQP不从API层进行限定而是直接定义网络交换的数据格式
• JMS
  • JMS即Java消息服务应用程序接口是一个java平台中关于面向消息中间件MOM的API用于在两个应用程序之间或分布式系统中发送消息进行异步通信。
• AMQP与JMS区别
  • JMS是定义了统一的接口来对消息操作进行统一AMQP是通过规定协议来统一数据交互的格式
  • JMS限定了必须使用java语言AMQP只是协议不规定实现方式因此是跨语言的
  • JMS规定了两种消息模式而AMQP的消息模式更加丰富
• 常见产品activeMQzeroMQRabbitMQRocketMQkafka

安装及配置RabbitMQ

目标按照文档在本机安装windows版本RabbitMQ并配置其用户和Virtual Hosts

• 分析
  1.安装erlang
  2.安装rabbitMQ
  3.安装RabbitMQ的图形管理界面插件
  4.创建管理用户
  5.创建虚拟主机Virtual Hosts
• 小结

安装上述的组件时候都需要使用以管理员身份运行。

3. 搭建RabbitMQ入门工程

目标搭建RabbitMQ入门工程并配置对应的maven依赖

分析

创建heima-rabbitmq的工程用于测试RabbitMQ的消息收发。添加用于操作RabbitMQ的依赖。

        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>

小结

使用IDEA创建maven工程使用了jdk1.8。在工程中的pom.xml文件中添加了上述的依赖。

4. 入门工程-生产者

目标编写消息生产者代码发送消息到队列

分析

入门工程生产者发送消息到RabbitMQ的队列simple_queue消费者可以从队列中获取消息。可以使用RabbitMQ的简单模式simple。

生产者实现发送消息的步骤

1. 创建连接工厂设置RabbitMQ的连接参数
2. 创建连接
3. 创建频道
4. 声明队列
5. 发送消息
6. 关闭资源

小结

package com.itheima.rabbitmq.simple;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * 简单模式发送消息
 */
public class Producer {
    static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws Exception {
        //1. 创建连接工厂设置RabbitMQ的连接参数
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机默认localhost
        connectionFactory.setHost("localhost");
        //连接端口默认5672
        connectionFactory.setPort(5672);
        //虚拟主机默认/
        connectionFactory.setVirtualHost("/itcast");
        //用户名默认guest
        connectionFactory.setUsername("heima");
        //密码默认guest
        connectionFactory.setPassword("heima");

        //2. 创建连接
        Connection connection = connectionFactory.newConnection();
        //3. 创建频道
        Channel channel = connection.createChannel();
        //4. 声明队列
        /**
         * 参数1队列名称
         * 参数2是否定义持久化队列消息会持久化保存在服务器上
         * 参数3是否独占本连接
         * 参数4是否在不使用的时候队列自动删除
         * 参数5其它参数
         */
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //5. 发送消息
        String message = "你好小兔纸。";

        /**
         * 参数1交换机名称如果没有则指定空字符串表示使用默认的交换机
         * 参数2路由key简单模式中可以使用队列名称
         * 参数3消息其它属性
         * 参数4消息内容
         */
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("已发送消息" + message);
        //6. 关闭资源
        channel.close();
        connection.close();
    }
}

在设置连接工厂的时候如果没有指定连接的参数则会有默认值可以去设置虚拟主机。

5. 入门工程-消费者

目标编写消息消费者代码从队列中接收消息并消费

分析

从RabbitMQ中队列与生产者发送消息时的队列一致simple_queue接收消息

实现消费者步骤

1. 创建连接工厂
2. 创建连接抽取一个获取连接的工具类
3. 创建频道
4. 声明队列
5. 创建消费者接收消息并处理消息
6. 监听队列

小结

package com.itheima.rabbitmq.simple;

import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * 简单模式消费者接收消息
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        //1. 创建连接工厂
        //2. 创建连接抽取一个获取连接的工具类
        Connection connection = ConnectionUtil.getConnection();
        //3. 创建频道
        Channel channel = connection.createChannel();
        //4. 声明队列
        /**
         * 参数1队列名称
         * 参数2是否定义持久化队列消息会持久化保存在服务器上
         * 参数3是否独占本连接
         * 参数4是否在不使用的时候队列自动删除
         * 参数5其它参数
         */
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        //5. 创建消费者接收消息并处理消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为" + envelope.getExchange());
                //消息id
                System.out.println("消息id为" + envelope.getDeliveryTag());
                //接收到的消息
                System.out.println("接收到的消息为" + new String(body, "utf-8"));
            }
        };
        //6. 监听队列
        /**
         * 参数1队列名
         * 参数2是否要自动确认设置为true表示消息接收到自动向MQ回复接收到了MQ则会将消息从队列中删除
         * 如果设置为false则需要手动确认
         * 参数3消费者
         */
        channel.basicConsume(Producer.QUEUE_NAME, true, defaultConsumer);
    }
}

需要持续监听队列消息所以不要关闭资源

6. 入门工程测试

目标启动消费者和生产者到RabbitMQ中查询队列并在消费者端IDEA控制台查看接收到的消息

分析

生产者发送消息到RabbitMQ队列simple_queue

消费者接收RabbitMQ队列消息

小结

简单模式生产者发送消息到队列中一个消费者从队列中接收消息。

在RabbitMQ中消费者只能从队列接收消息。

如果接收消息的消费者在同一个队列中有两个或多个时消息是如何分配的

7. Work queues工作队列模式

目标编写生产者、消费者代码并测试了解Work queues工作队列模式的特点

分析

工作队列模式在同一个队列中可以有多个消费者消费者之间对于消息的接收是竞争关系。

生产者发送30个消息

消费者创建两个消费者监听同一个队列查看两个消费者的接收消息是否存在重复。

小结

工作队列模式一个消息只能被一个消费者接收其它消费者是不能接收到同一条消息的。

应用场景可以在消费者端处理任务比较耗时的时候添加对同一个队列的消费者来提高任务处理能力。

8. 订阅模式类型说明

目标说出订阅模式中的Exchange交换机作用以及交换机的三种类型

小结

订阅模式与前面的两种模式比较多了一个角色Exchange交换机接收生产者发送的消息并决定如何投递消息到其绑定的队列消息的投递决定于交换机的类型。

交换机类型广播fanout、定向direct、通配符topic

交换机只做消息转发自身不存储数据。

9. Publish/Subscribe发布与订阅模式

目标编写生产者、消费者代码并测试了解Publish/Subscribe发布与订阅模式的特点

分析

发布与订阅模式特点一个消息可以被多个消费者接收其实是使用了订阅模式交换机类型为fanout广播

• 生产者发送10个消息
  1. 创建连接
  2. 创建频道
  3. 声明交换机fanout
  4. 声明队列
  5. 队列绑定到交换机
  6. 发送消息
  7. 关闭资源
• 消费者至少两个消费者
  1. 创建连接
  2. 创建频道
  3. 声明交换机
  4. 声明队列
  5. 队列绑定到交换机
  6. 创建消费者
  7. 监听队列

小结

发布与订阅模式一个消息可以被多个消费者接收一个消费者对于的队列该队列只能被一个消费者监听。使用了订阅模式中交换机类型为广播。

10. Routing路由模式

目标编写生产者、消费者代码并测试了解Routing路由模式的特点

分析

生产者发送两条消息路由key分别为insert、update

消费者创建两个消费者监听的队列分别绑定路由key为insert、update

1. 消息中路由key为insert的会被绑定路由key为insert的队列接收并被其监听的消费者接收、处理
2. 消息中路由key为update的会被绑定路由key为update的队列接收并被其监听的消费者接收、处理

小结

Routing 路面模式要求队列绑定到交换机的时候指定路由key消费发送时候需要携带路由key只有消息的路由key与队列路由key完全一致才能让该队列接收到消息。

11. Topics通配符模式

目标编写生产者、消费者代码并测试了解Topics通配符模式的特点

分析

• 生产者发送包含有item.insert、item.updateitem.delete的3中路由key消息
• 消费者1监听的队列绑定到交换机的路由key为item.updateitem.delete
• 消费者2监听的队列绑定到交换机的路由key为item.*

小结

Topics通配符模式可以根据路由key将消息传递到对应路由key的队列队列绑定到交换机的路由key可以有多个通配符模式中路由key可以使用 * 和 # 使用了通配符模式之后对于路由Key的配置更加灵活。

12. RabbitMQ模式总结

目标对比总结RabbitMQ的5种模式特征

小结

• 不直接Exchange交换机默认交换机
  1. simple简单模式一个生产者生产一个消息到一个队列被一个消费者接收
  2. work工作队列模式生产者发送消息到一个队列中然后可以被多个消费者监听该队列一个消息只能被一个消费者接收消费者之间是竞争关系
• 使用Exchange交换机订阅模式交换机广播fanout、定向direct、通配符topic
  1. 发布与订阅模式使用了fanout广播类型的交换机可以将一个消息发送到所有绑定了该交换机的队列
  2. 路由模式使用了direct定向类型的交换机消费会携带路由key交换机根据消息的路由key与队列的路由key进行对比一致的话那么该队列可以接收到消息
  3. 通配符模式使用了topic通配符类型的交换机消费会携带路由key* #交换机根据消息的路由key与队列的路由key进行对比匹配的话那么该队列可以接收到消息

13. 创建SpringBoot整合RabbitMQ的两个工程

目标创建springboot-rabbitmq-producer工程用于生产消息创建springboot-rabbitmq-consumer工程用于接收消息

分析

Spring Boot提供了对于AMQP的整合可以使用RabbitTemplate发送消息可以使用@RabbitListener注解接收消息。

生产者工程springboot-rabbitmq-producer发送消息

1. 创建工程
2. 添加依赖spring-boot-stater-amqpspring-boot-starter-test
3. 创建启动引导类
4. 添加配置文件application.yml

消费者工程springboot-rabbitmq-consumer接收消息

1. 创建工程
2. 添加依赖spring-boot-stater-amqp
3. 创建启动引导类
4. 添加配置文件application.yml

小结

可以使用插件自动生产Spring Boot工程的启动引导类Application.java和配置文件application.yml

14. 配置生产者工程

目标配置springboot-rabbitmq-producer工程的RabbitMQ一个交换机、队列并绑定

分析

使用通配符模式将队列绑定到交换机topic时需要指定路由keyitem.#

1. 配置RabbitMQ的连接参数主机、连接端口、虚拟主机、用户名、密码
2. 声明交换机、队列并将队列绑定到交换机指定的路由keyitem.#

小结

• 配置application.yml文件

spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /itcast
    username: heima
    password: heima

• 配置交换机、队列和绑定创建一个配置类

@Configuration
public class RabbitMQConfig {
    //交换机名称
    public static final String ITEM_TOPIC_EXCHANGE = "item_topic_exchange";

    //队列名称
    public static final String ITEM_QUEUE = "item_queue";

    //声明交换机
    @Bean("itemTopicExchange")
    public Exchange topicExchange(){
        return ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build();
    }

    //声明队列
    @Bean("itemQueue")
    public Queue itemQueue(){
        return QueueBuilder.durable(ITEM_QUEUE).build();
    }

    //将队列绑定到交换机
    @Bean
    public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue,
                                     @Qualifier("itemTopicExchange")Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs();
    }
}

15. 配置消费者工程

目标配置springboot-rabbitmq-consumer工程的RabbitMQ编写消息监听器接收消息

分析

1. 配置application.yml文件设置RabbitMQ的连接参数
2. 编写消息监听器接收队列item_queue消息可以使用注解@RabbitListener接收队列消息

小结

• 配置application.yml文件与生产者工程一致
• 编写监听器类

@Component
public class MyListener {

    /**
     * 接收队列消息
     * @param message 接收到的消息
     */
    @RabbitListener(queues = "item_queue")
    public void myListener1(String message){
        System.out.println("消费者接收到消息" + message);
    }
}

接收消息的队列名称要与生产者发送消息时的队列名称一致

16. 测试消息发送和接收

目标生产者编写测试类RabbitMQTest发送消息到交换机和特定的路由item.insertitem.updateitem.delete

分析

生产者编写测试类RabbitMQTest利用RabbitTemplate发送3条消息这3条消息的路由key分别是item.insertitem.updateitem.delete

消费者在IDEA控制台查看是否能接收到符合路由key的消息

小结

编写测试类如下

package com.itheima.rabbitmq;

import com.itheima.rabbitmq.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMQTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void test(){
        rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE,
                "item.insert", "商品新增,路由Key为item.insert");
        rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE,
                "item.update", "商品新增,路由Key为item.update");
        rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE,
                "item.delete", "商品新增,路由Key为item.delete");
        rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE,
                "a.item.delete", "商品新增,路由Key为a.item.delete");
    }
}

先启动测试类进行声明交换机、队列和绑定之后再启动消费者工程接收消息。