RabbitMQ

阅读量

1.MQ引言

1.1 什么是MQ

MQ(Message Quene) : 翻译为 消息队列，通过典型的 生产者和消费者模型，生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，轻松的实现系统间解耦。别名为 消息中间件 通过利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。

1.2 MQ有哪些

当今市面上有很多主流的消息中间件，如老牌的ActiveMQ、RabbitMQ，Kafka，阿里巴巴自主开发RocketMQ等。

1.3 不同MQ特点

1
2
3
4
5
6
7
8
9
10
11
12
13
# 1.ActiveMQ
    ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。它是一个完全支持JMS规范的的消息中间件。丰富的API,多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件,在中小型企业颇受欢迎!

# 2.Kafka
    Kafka是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务（大数据效率第一，丢失一两个信息不影响最后的结果）。

# 3.RocketMQ
    RocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。
    RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

# 4.RabbitMQ
    RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
    AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。

RabbitMQ比Kafka可靠，Kafka更适合IO高吞吐的处理，一般应用在大数据日志处理或对实时性（少量延迟），可靠性（少量丢数据）要求稍低的场景使用，比如ELK日志收集。

2.RabbitMQ 的引言

2.1 RabbitMQ

基于AMQP协议，erlang语言开发，是部署最广泛的开源消息中间件,是最受欢迎的开源消息中间件之一。

官网: https://www.rabbitmq.com/

官方教程: https://www.rabbitmq.com/#getstarted

1
2
3
4
5
6
7
8
9
# AMQP 协议
    AMQP（advanced message queuing protocol）在2003年时被提出，最早用于解决金融领不同平台之间的消息传递交互问题。顾名思义，AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。
    这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。
    以下是AMQP协议模型:
    	生产者：发布消息到RabbitMQ中的Exchange。
    	消费者：监听RabbitMQ中的Queue中的消息。
    	交换机：和生产者建立连接，并接收生产者的消息。并将消息分发给指定的消息队列。
    	队列：接收交换机发的消息，并和消费者进行交互。
    	路由(Routes)：交换机以什么样的策略将消息发布到消息队列。

image-20200311182438041

2.2 RabbitMQ 的安装

2.2.1 下载

官网下载地址: https://www.rabbitmq.com/download.html
image-20190925220115235

2.2.2 下载的安装包!

image-20190925220343521

注意:这里的安装包是centos7安装的包

2.2.3 安装步骤

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 1.将rabbitmq安装包上传到linux系统中
	erlang-22.0.7-1.el7.x86_64.rpm
	rabbitmq-server-3.7.18-1.el7.noarch.rpm

# 2.安装Erlang依赖包
	rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm

# 3.安装RabbitMQ安装包(需要联网)
	yum install -y rabbitmq-server-3.7.18-1.el7.noarch.rpm
	注意:默认安装完成后配置文件模板在:/usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example目录中,需要将配置文件复制到/etc/rabbitmq/目录中,并修改名称为rabbitmq.config
# 4.复制配置文件
	cp /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

# 5.查看配置文件位置
	ls /etc/rabbitmq/rabbitmq.config

# 6.修改配置文件(参见下图:)
	vim /etc/rabbitmq/rabbitmq.config

image-20190925222230260
将上图中配置文件中红色部分去掉%%,以及最后的,逗号。修改为下图:
image-20190925222329200

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# 7.执行如下命令,启动RabbitMQ中的插件管理  
	rabbitmq-plugins enable rabbitmq_management
	
	出现如下说明: 
		Enabling plugins on node rabbit@localhost:
    rabbitmq_management
    The following plugins have been configured:
      rabbitmq_management
      rabbitmq_management_agent
      rabbitmq_web_dispatch
    Applying plugin configuration to rabbit@localhost...
    The following plugins have been enabled:
      rabbitmq_management
      rabbitmq_management_agent
      rabbitmq_web_dispatch

    set 3 plugins.
    Offline change; changes will take effect at broker restart.

# 8.启动RabbitMQ的服务    
	systemctl start rabbitmq-server    （启动命令）
	systemctl restart rabbitmq-server  （重启命令）
	systemctl stop rabbitmq-server     （关闭命令） 
	
   
# 9.查看服务状态(见下图:)   
	systemctl status rabbitmq-server
  ● rabbitmq-server.service - RabbitMQ broker
     Loaded: loaded (/usr/lib/systemd/system/rabbitmq-server.service; disabled; vendor preset: disabled)
     Active: active (running) since 三 2019-09-25 22:26:35 CST; 7s ago
   Main PID: 2904 (beam.smp)
     Status: "Initialized"
     CGroup: /system.slice/rabbitmq-server.service
             ├─2904 /usr/lib64/erlang/erts-10.4.4/bin/beam.smp -W w -A 64 -MBas ageffcbf -MHas ageffcbf -
             MBlmbcs...
             ├─3220 erl_child_setup 32768
             ├─3243 inet_gethost 4
             └─3244 inet_gethost 4
      .........

image-20190925222743776

1
2
3
4
5
6
7
8
# 10.关闭防火墙服务（或者打开15672端口）
	systemctl disable firewalld
    Removed symlink /etc/systemd/system/multi-user.target.wants/firewalld.service.
    Removed symlink /etc/systemd/system/dbus-org.fedoraproject.FirewallD1.service.
	systemctl stop firewalld   

# 11.访问web管理界面
	http://IP地址:15672/

image-20190926194738708

1
2
3
# 12.登录管理界面
	username:  guest
	password:  guest

image-20190926194954822

3. RabiitMQ 配置

3.1RabbitMQ 管理命令行

1
2
3
4
5
6
7
8
9
10
11
12
13
# 1.服务启动相关
	systemctl start rabbitmq-server    （启动命令）
	systemctl restart rabbitmq-server  （重启命令）
	systemctl stop rabbitmq-server     （关闭命令）
	systemctl status rabbitmq-server   （查看状态命令）

# 2.管理命令行  用来在不使用web管理界面情况下命令操作RabbitMQ
	rabbitmqctl  help  可以查看更多命令

# 3.插件管理命令行
	rabbitmq-plugins list        （查看所有插件）
	rabbitmq-plugins enable xxx  （启动一个或多个插件）
	rabbitmq-plugins disable xxx （禁用一个或多个插件）

3.2 web管理界面介绍

3.2.1 overview概览

image-20191126162026720

  • connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况。
  • channels：通道。建立连接后，会形成通道。消息的投递获取依赖通道。
  • Exchanges：交换机。用来实现消息的路由。
  • Queues：队列，即消息队列。消息存放在队列中，等待消费，消费后被移除队列。
  • Admin：对用户进行管理。

3.2.2 Admin用户和虚拟主机管理

1. 添加用户

image-20191126162617280

上面的Tags选项，其实是指定用户的角色，可选的有以下几个：

  • 超级管理员(administrator)
    可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。
  • 监控者(monitoring)
    可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
  • 策略制定者(policymaker)
    可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
  • 普通管理者(management)
    仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。
  • 其他
    无法登陆管理控制台，通常就是普通的生产者和消费者。
2. 创建虚拟主机
1
2
3
4
# 虚拟主机
	为了让各个用户可以互不干扰的工作，RabbitMQ添加了虚拟主机（Virtual Hosts）的概念。其实就是一个独立的访问路径，不同用户使用不同路径，各自有自己的队列、交换机，互相不会影响。
	生产者A连接到虚拟主机A1，才能发布消息。消费者B链接到虚拟主机A1，才能接收到生产者A发布的消息。
	相当于数据库中的库。

image-20191126163023153

3. 绑定虚拟主机和用户

创建好虚拟主机，我们还要给用户添加访问权限：
点击添加好的虚拟主机：
image-20191126163506795
进入虚拟机设置界面:
image-20191126163631889

4.RabbitMQ 的第一个程序

4.0 AMQP协议的回顾

image-20200312140114784

4.1 RabbitMQ支持的消息模型

image-20191126165434784
image-20191126165459282

4.2 引入依赖

1
2
3
4
5
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.7.2</version>
</dependency>

4.3 第一种模型(直连)

image-20191126165840602

在上图的模型中，有以下概念：

  • P：生产者，也就是要发送消息的程序。
  • C：消费者：消息的接受者，会一直等待消息到来。
  • queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
1. 消息生产者
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
// 生产消息
@Test
public void sendMessageTest() throws Exception {

    // 创建连接MQ的连接工厂对象
    ConnectionFactory connectionFactory = new ConnectionFactory();

    // 配置连接工厂
    connectionFactory.setHost("10.15.0.9"); // ip
    connectionFactory.setPort(5672);        // 端口号
    connectionFactory.setVirtualHost("/ems"); // 设置要连接哪个虚拟主机
    connectionFactory.setUsername("ems");   // 要连接的虚拟主机名
    connectionFactory.setPassword("123");   // 要连接的虚拟主机密码

    // 获取连接对象
    Connection connection = connectionFactory.newConnection();

    // 通过连接，获取通道
    Channel channel = connection.createChannel();

    /**
     * 通道绑定对应的消息队列
     * 参数1: 用来声明通道对应的队列名称。如果此队列不存在，则会自动创建。
     * 参数2: 定义此队列是否要持久化（注意不是队列中的消息持久化）。true：持久化。
     * 参数3: 是否独占队列。true：独占，代表当前队列只允许当前的连接可用。
     * 参数4: 队列中的消息被消费完后是否自动删除队列。true：自动删除
     * 参数5: 额外附加属性
     */
    channel.queueDeclare("hello", true, false, false, null);

    /**
     * 发布消息：
     * 参数1: 交换机名（本案例直接通过生产者到队列，没有交换机）。
     * 参数2: 队列名称，发布消息到具体的哪个队列。
     * 参数3: 发布消息时的一些属性。
     	如消息持久化参数：MessageProperties.PERSISTENT_TEXT_PLAIN
     * 参数4: 发布消息的内容。（要求是字节数组）
     */
    channel.basicPublish("","hello", null,"hello rabbitmq".getBytes());

    // 关闭资源
    channel.close(); 
    connection.close();
}
2. 消息消费者
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
// 消费消息(在main中测试，是为了保证消费者一直处于监听状态，不会自动停止。)
public void main(String[] args) throws Exception {
    // 消费者要和生产者绑定同一个虚拟主机。并找到相应的队列。
    // 创建连接工厂
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("10.15.0.9");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/ems");
    connectionFactory.setUsername("ems");
    connectionFactory.setPassword("123");

    // 创建连接对象
    Connection connection = connectionFactory.newConnection();

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

    // 通道绑定队列
    channel.queueDeclare("hello", true, false, false, null);

    /**
     * 开启监听Queue：消费消息
     * 参数1：队列名称：消费哪个队列中的消息。
     * 参数2：是否开启 消息的自动确认机制ACK。true：接收到消息后，会立刻告诉RabbitMQ
     * 参数3：消费消息时的回调接口。
     */
    channel.basicConsume("hello",true, new DefaultConsumer(channel){
        // 消费消息时的回调方法
        @Override
        public void handleDelivery(String consumerTag,
                                   Envelope envelope,
                                   AMQP.BasicProperties properties,
                                   byte[] body) throws IOException {
            // body：从消息队列中取出的消息
            System.out.println("获取的消息：" + new String(body));
        }
    });

    // 关闭资源（如果不关闭，就会一直监听队列）
    channel.close();
    connection.close();
}
3. 封装连接工具类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
// 简单地封装以下 供测试使用
public class RabbitMQUtil {

    // 类加载时就创建，且每次调用方法，无需再创建新的。
    public static ConnectionFactory connectionFactory;

    // 类加载的时候执行，且只执行一次
    // 一般用来写一些赋值操作（初始化）
    static {
        //创建连接MQ的连接工厂对象
        connectionFactory = new ConnectionFactory();

        // 配置连接工厂
        connectionFactory.setHost("10.15.0.9"); // ip
        connectionFactory.setPort(5672);        // 端口号
        connectionFactory.setVirtualHost("/ems"); // 设置要连接哪个虚拟主机
        connectionFactory.setUsername("ems");   // 要连接的虚拟主机名
        connectionFactory.setPassword("123");   // 要连接的虚拟主机密码
    }

    // 提供获取连接对象的方法
    public static Connection getConnection() {
        try {
            // 获取连接对象
            return connectionFactory.newConnection();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    // 关闭通道和连接
    public static void closeConnAndChannel(Channel ch, Connection conn){
        try {
            if(ch != null) ch.close();
            if (conn != null) conn.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}


// 以后创建通道方法
Channel channel = RabbitMQUtil.getConnection().createChannel();

4.4 第二种模型(work quene)

Work queues，也被称为（Task queues），任务模型。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。
image-20200314221002008

角色：

  • P：生产者：任务的发布者
  • C1：消费者-1，领取任务并且完成任务，假设完成速度较快
  • C2：消费者-2：领取任务并完成任务，假设完成速度慢
1. 生产者
1
2
3
4
5
6
channel.queueDeclare("hello", true, false, false, null);

// 往hello队列中 发布10个消息
for (int i = 0; i < 10; i++) {
  channel.basicPublish("", "hello", null, (i + "====>:我是消息").getBytes());
}
2.消费者-1
1
2
3
4
5
6
7
8
9
10
11
12
13
channel.queueDeclare("hello", true, false, false, null);

channel.basicConsume("hello", true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
    System.out.println("消费者1: " + new String(body));
      
  }
});
3.消费者-2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
channel.queueDeclare("hello", true, false, false, null);

channel.basicConsume("hello", true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
    try {
      Thread.sleep(1000);   // 处理消息比较慢 一秒处理一个消息
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("消费者2: " + new String(body));  
      
  }
});
4.测试结果

image-20200314223242058
image-20200314223302207

总结:在不做任何设置的情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。

平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。

上面遇到的问题：
消费者2的处理速度比消费者1慢，而我们采用的是平均分配，这就导致消费者2拖慢了系统的允许速度。
我们希望处理快的能多处理一些消息，处理慢的就少处理一些，能者多劳。这就要使用消息种动确认机制。

5.消息自动确认机制

image-20200314221002008

假设消费者2已经被系统分配了5个消息，这5个消息就会被RabbitMQ标记为清除。

当消费到第3个消息的时候，消费者自己宕机了，那么还剩2个消息就丢失了。

我们希望未被处理的消息能够交给消费者1进行处理，而不是丢失。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/*
	步骤：
	设置通道一次只能消费一个消息。
	关闭消息的自动确认,开启手动确认消息。
*/

// 指定当前消费者，一次只消费一条未确认的消息
channel.basicQos(1); 

// 参数2:关闭自动确认消息
channel.basicConsume("hello", false, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
    System.out.println("消费者1: " + new String(body));
    channel.basicAck(envelope.getDeliveryTag(), false); // 手动确认消息
      
  }
});

结果：

image-20200314230412178
image-20200314230423280

4.5 第三种模型(fanout)

fanout 扇出 也称为广播

image-20191126213115873

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

  • 可以有多个消费者。
  • 每个消费者有自己的queue（队列）。
  • 每个队列都要绑定到Exchange（交换机）。
  • 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
  • 交换机把消息发送给 绑定过此交换机的所有队列。
  • 队列的消费者都能拿到消息。实现一条消息被多个消费者消费。
1. 生产者
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/*
	绑定交换机
	参数1：交换机名称(可自定义)
	参数2：交换机类型(fanout：代表广播类型)
	广播：一条消息多个消费者同时消费
*/
channel.exchangeDeclare("logs", "fanout"); 

/*
   绑定队列：将队列绑定到交换机exchange（队列名, 交换机名, 路由key）
   一般是在生产者上将队列绑定到交换机，那么消费者就直接消费队列就行了。
   这里采用的是消费者声明零时队列，并进行绑定。生产者不进行这一步操作。
*/
// channel.queueBind("queue", "logs", "");

/*
	发布消息
	参数1：要发布到哪个交换机
	参数2：routingKey，在广播里，路由key是无任何意义的，所以空着。
*/
channel.basicPublish("logs", "", null, "hello".getBytes());

// 省略释放资源操作。
2. 消费者-1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 绑定交换机（可以用声明交换机的方式来绑定）
channel.exchangeDeclare("logs", "fanout");
// 创建临时队列（针对广播，创建临时队列就行，）
String queue = channel.queueDeclare().getQueue();
// 将临时队列绑定到交换机exchange（队列名, 交换机名, 路由key）
channel.queueBind(queue, "logs", "");
// 消费消息
channel.basicConsume(queue, true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
       System.out.println("消费者1: " + new String(body));
      
  }
});
3. 消费者-2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 绑定交换机（可以用声明交换机的方式来绑定）
channel.exchangeDeclare("logs", "fanout");
// 创建临时队列（针对广播，创建临时队列就行，）
String queue = channel.queueDeclare().getQueue();
// 将临时队列绑定到交换机exchange（队列名, 交换机名, 路由key）
channel.queueBind(queue, "logs", "");
// 消费消息
channel.basicConsume(queue, true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
       System.out.println("消费者2: " + new String(body));
      
  }
});
4.消费者-3
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 绑定交换机（可以用声明交换机的方式来绑定）
channel.exchangeDeclare("logs", "fanout");
// 创建临时队列（针对广播，创建临时队列就行，）
String queue = channel.queueDeclare().getQueue();
// 将临时队列绑定到交换机exchange（队列名, 交换机名, 路由key）
channel.queueBind(queue, "logs", "");
// 消费消息
channel.basicConsume(queue, true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
       System.out.println("消费者3: " + new String(body));
      
  }
});
5. 测试结果

image-20200315180653207

image-20200315180708489

image-20200315180728035

4.6 第四种模型(Routing)

4.6.1 Routing 之订阅模型-Direct(直连)

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

在Direct模型下：

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

流程:

image-20191126220145375

图解：

  • P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
  • X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列。
  • C1：消费者，其所在队列指定了需要routing key 为 error 的消息。
  • C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息。
1. 生产者
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/*
	声明交换机
	参数1：交换机名称(可自定义)
	参数2：交换机类型(direct：路由模式)
*/
channel.exchangeDeclare("logs_direct", "direct");

// 路由key：可自定义。这里以error为例。
String key = "error";

// 发布消息
channel.basicPublish("logs_direct", key, null, 
                     ("指定的route key" + key + "的消息").getBytes());

// 省略关闭资源
2.消费者-1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// 绑定交换机（可以用声明交换机的方式来绑定）
channel.exchangeDeclare("logs_direct", "direct");

// 创建临时队列（针对此项目，创建临时队列就行，）
String queue = channel.queueDeclare().getQueue();

/*
	将临时队列绑定到交换机exchange
	参数（队列名, 交换机名, 路由key）
	下面三行表示，此消费者可接收路由key为error、info、warn的消息。
*/
channel.queueBind(queue, "logs_direct", "error");
channel.queueBind(queue, "logs_direct", "info");
channel.queueBind(queue, "logs_direct", "warn");

// 消费消息
channel.basicConsume(queue, true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
      System.out.println("消费者1: " + new String(body));
      
  }
});
3.消费者-2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 绑定交换机（可以用声明交换机的方式来绑定）
channel.exchangeDeclare("logs_direct", "direct");

// 创建临时队列（针对此项目，创建临时队列就行，）
String queue = channel.queueDeclare().getQueue();

// 将临时队列绑定到交换机exchange（队列名, 交换机名, 路由key）
channel.queueBind(queue, "logs_direct", "error");

// 消费消息
channel.basicConsume(queue, true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
      System.out.println("消费者2: " + new String(body));
      
  }
});
4.测试生产者发送Route key为error的消息时

image-20200316102613933

image-20200316102627912

5.测试生产者发送Route key为info的消息时（此时消息队列中已经有error的消息了）

image-20200316102925740

image-20200316102947326

4.6.2 Routing 之订阅模型-Topic

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

试想如果没有通配符，代码冗余情况： 

1
2
3
channel.queueBind(queue, "logs_direct", "error2");
channel.queueBind(queue, "logs_direct", "error3");
channel.queueBind(queue, "logs_direct", "error4");

image-20191127121900255

1
2
3
4
5
6
7
8
9
这种模型Routingkey一般都是由一个或多个单词组成。
多个单词之间以”.”分割，例如： item.insert

# 统配符
	* ：匹配1个词
	# ：匹配一个或多个词
# 如: 有这2个Routingkey -> audit.irs.corporate、audit.irs
    audit.*    只能匹配 audit.irs
	audit.#    匹配 audit.irs.corporate、audit.irs
1.生产者
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/*
	声明交换机
	参数1：交换机名称(可自定义)
	参数2：交换机类型(topic：使用动态路由(通配符方式))
*/
channel.exchangeDeclare("topics", "topic");

// 动态路由key
String routekey = "user.save"; 

// 发布消息
channel.basicPublish("topics", 
                     routekey, 
                     null, 
                     ("这是路由中的动态订阅模型,route key: ["+routekey+"]").getBytes());
2.消费者-1

Routing Key中使用*通配符方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 声明交换机
channel.exchangeDeclare("topics", "topic");

// 创建临时队列
String queue = channel.queueDeclare().getQueue();

// 绑定队列与交换机,并设置获取交换机中动态路由
channel.queueBind(queue, "topics", "user.*");

// 消费消息
channel.basicConsume(queue, true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
      System.out.println("消费者1: " + new String(body));
      
  }
});
3.消费者-2

Routing Key中使用#通配符方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
//声明交换机
channel.exchangeDeclare("topics", "topic");

//创建临时队列
String queue = channel.queueDeclare().getQueue();

// 绑定队列与交换机,并设置获取交换机中动态路由
channel.queueBind(queue, "topics", "user.#");

// 消费消息
channel.basicConsume(queue, true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
      System.out.println("消费者2: "+new String(body));
      
  }
});
4.测试结果

image-20200316113935785

image-20200316114000459

5. SpringBoot中使用RabbitMQ

5.0 搭建初始环境

1. 引入依赖
1
2
3
4
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2. 配置配置文件
1
2
3
4
5
6
7
spring:
  rabbitmq:
    host: 10.15.0.9
    port: 5672
    username: ems
    password: 123
    virtual-host: /ems

5.1 第一种hello world模型使用

  1. 生产者

    RabbitTemplate 用来简化操作，使用时候直接在项目中注入即可使用。

    1
    2
    3
    4
    5
    6
    7
    8
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testHello(){
        // 转换and发送：发送内容"hello world"（发送内容的类型可为任意类型）
        rabbitTemplate.convertAndSend("hello", "hello world");
    }
  2. 消费者
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    /*
      消费者监听hello队列，若没有hello队列，则自动创建。
      @Queue：设置消息队列属性（消息队列名、是否独占、是否自动删除等）。
    */
    @Component
    @RabbitListener(queuesToDeclare = @Queue("hello"))
    public class HelloCustomer {

        /*
           @RabbitHandler：代表此方法是从队列中取出消息时的回调方法
           message：从队列中取出的消息
        */
        @RabbitHandler
        public void receive1(String message){
            System.out.println("message = " + message);
        }
    }

5.2 第二种work模型使用

  1. 生产者
    1
    2
    3
    4
    5
    6
    7
    8
    9
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testWork(){
      for (int i = 0; i < 10; i++) {
        rabbitTemplate.convertAndSend("work", "hello work!");
      }
    }
  2. 消费者
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    @Component
    public class WorkCustomer {
        
        @RabbitListener(queuesToDeclare = @Queue("work"))
        public void receive1(String message){
            System.out.println("work message1 = " + message);
        }

        @RabbitListener(queuesToDeclare = @Queue("work"))
        public void receive2(String message){
            System.out.println("work message2 = " + message);
        }
    }

    说明:默认在Spring AMQP实现中Work这种方式就是公平调度,如果需要实现能者多劳需要额外配置。

5.3 Fanout 广播模型

  1. 生产者
    1
    2
    3
    4
    5
    6
    7
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testFanout() throws InterruptedException {
        rabbitTemplate.convertAndSend("logs", "", "这是日志广播");
    }
  1. 消费者
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    @Component
    public class FanoutCustomer {

        @RabbitListener(bindings = @QueueBinding(
                value = @Queue, //创建临时队列
                // 绑定交换机，并指定交换机类型
                exchange = @Exchange(name="logs", type = "fanout") 
        ))
        public void receive1(String message){
            System.out.println("message1 = " + message);
        }
        

        @RabbitListener(bindings = @QueueBinding(
                value = @Queue, 
                exchange = @Exchange(name="logs",type = "fanout")  
        ))
        public void receive2(String message){
            System.out.println("message2 = " + message);
        }
    }

5.4 Route 路由模型

  1. 生产者
    1
    2
    3
    4
    5
    6
    7
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testDirect(){
        rabbitTemplate.convertAndSend("directs", "error", "error 的日志信息");
    }
  2. 消费者
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    @Component
    public class DirectCustomer {

        @RabbitListener(bindings ={
                @QueueBinding(
                        value = @Queue(),
                    // 指定路由key
                        key = {"info", "error"}, 
                        exchange = @Exchange(type = "direct", name = "directs")
                )})
        public void receive1(String message){
            System.out.println("message1 = " + message);
        }

        
        @RabbitListener(bindings ={
                @QueueBinding(
                        value = @Queue(),
                        key = {"error"},
                        exchange = @Exchange(type = "direct", name = "directs")
                )})
        public void receive2(String message){
            System.out.println("message2 = " + message);
        }
    }

5.5 Topic 订阅模型(动态路由模型)

  1. 生产者
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    @Autowired
    private RabbitTemplate rabbitTemplate;

    //topic
    @Test
    public void testTopic(){
      rabbitTemplate.convertAndSend("topics", 
                                    "user.save.findAll",
                                    "user.save.findAll 的消息");
    }
  1. 消费者
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    @Component
    public class TopCustomer {
        @RabbitListener(bindings = {
                @QueueBinding(
                        value = @Queue,
                        key = {"user.*"},
                        exchange = @Exchange(type = "topic", name = "topics")
                )
        })
        public void receive1(String message){
            System.out.println("message1 = " + message);
        }
        
        

        @RabbitListener(bindings = {
                @QueueBinding(
                        value = @Queue,
                        key = {"user.#"},
                        exchange = @Exchange(type = "topic", name = "topics")
                )
        })
        public void receive2(String message){
            System.out.println("message2 = " + message);
        }
     }

6. 消息可靠性

如何保证生产者生产的消息能被消费者100%消费？

问题：

  1. 如果消息已经到达RabbitMQ，但RabbitMQ宕机了，消息是不是就丢了？不会，RabbitMQ的Queue有持久化机制。
  2. 消费者在消费消息时，如果执行到一半，消费者宕机了怎么办？手动ACK。
  3. 如果生产者发送消息时，由于网络原因，导致消息没发送到RabbitMQ，该怎么办？

6.1 Confirm机制

解决第3问的办法：RabbitMQ提供了事务操作和Confirm操作。

RabbitMQ的事务：事务可保证消息100%传递。可以通过事务的回滚去记录日志，后面定时再次发送当前消息。但是事务的操作，效率太低。（加事务效率要降低100倍以上，所以不建议用）

Confirm：确认机制。效率比事务高。保证把消息到达Exchange。不能保证消息被Exchange分发到指定Queue

1.普通Confirm方式：发送一个消息的时候，等待对方告诉自己是否发送成功。

2.批量Confirm方式：发送多个消息的时候使用。

3.异步Confirm方式：推荐使用。

6.1.1 普通Confirm方式

生产者：开启confirm、waitForConfirms判断
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 常规操作
@Test
public void sendMessageTest() throws Exception {

    channel.queueDeclare("hello", true, false, false, null);
    
    // 1.发送消息之前开启confirm
    channel.confirmSelect();

    // 发布消息：
    channel.basicPublish("","hello", null,"hello rabbitmq".getBytes());
    
    // 2.发布消息之后，用confirm进行判断
    if(channel.waitForConfirms()) {
        System.out.println("消息发送成功");
    }else {
        System.out.println("消息发送失败");
        // 进行后续操作
    }

    // 省略关闭资源
}

6.1.2 批量Confirm方式

生产者：开启confirm、waitForConfirmsOrDie反馈
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 常规操作
@Test
public void sendMessageTest() throws Exception {

    channel.queueDeclare("hello", true, false, false, null);
    
    // 1.发送消息之前开启confirm
    channel.confirmSelect();

    // 发布消息：
    for (int i = 0; i < 10; i++) {
        String msg = "hello" + i;
        channel.basicPublish("","hello", null, msg.getBytes());
    }
    
    
    // 2.发送的全部消息，有一个失败的时候，就全部失败，并抛出异常
    channel.waitForConfirmsOrDie();

    // 省略关闭资源
}

6.1.3 异步Confirm方式

生产者：开启confirm、addConfirmListener
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
// 常规操作
@Test
public void sendMessageTest() throws Exception {

    channel.queueDeclare("hello", true, false, false, null);
    
    // 1.发送消息之前开启confirm
    channel.confirmSelect();

    // 发布消息：
    for (int i = 0; i < 10; i++) {
        String msg = "hello" + i;
        channel.basicPublish("","hello", null, msg.getBytes());
    }
    
    
    // 2.异步方式
    channel.addConfirmListener(new ConfirmListener() {
        // 发送成功的回调方法
        @Override
        public void handleAck(long deliveryTag, boolean multiple) throws IOException {
            System.out.println("消息发送成功");
            System.out.println("消息标识：" + deliveryTag);
            System.out.println("是否是批量操作：" + multiple);
        }
        
        // 发送失败的回调方法
        @Override
        public void handleNack(long deliveryTag, boolean multiple) throws IOException {
            System.out.println("消息发送失败");
            System.out.println("消息标识：" + deliveryTag);
            System.out.println("是否是批量操作：" + multiple);
        }
    });

    // 省略关闭资源
}

6.2 Return机制

Confirm机制只保证把消息发送到Exchange上。而消费者监听的是消息队列。且Exchange不能持久化消息，Queue是可以持久化消息。

若Exchange发消息发给队列的过程中出现了错误，该怎么办呢？

Return机制：采用Return机制来监听消息是否从Exchange送到了指定的Queue。

6.2.1 使用Return机制

添加Return监听器、配置Return回调方法，开启Return机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
// 常规操作
@Test
public void sendMessageTest() throws Exception {

    channel.queueDeclare("hello", true, false, false, null);
    
    // 1.添加Return监听器、配置Return回调方法。
    channel.addReturnListener(new ReturnListener() {
        // 回调函数：当消息没有送到Queue时，会执行此回调函数。
        @Override
        public void handleReturn(int replyCode, 
                                 String replyText, 
                                 String exchange, 
                                 String routingKey, 
                                 AMQP.BasicProperties properties, 
                                 byte[] body) throws IOException {
            
            System.out.println(new String(body) + "未送到Queue中");
        }
    });
    
    // 发送消息之前开启confirm
    channel.confirmSelect();

    // 2.开启Return机制。添加参数（第3个），并设为true。
    // 发布消息：
    channel.basicPublish("","hello", true, null, "hello".getBytes());
    
    // 发布消息之后，用confirm进行判断
    if(channel.waitForConfirms()) {
        System.out.println("消息发送成功");
    }else {
        System.out.println("消息发送失败");
        // 进行后续操作
    }

    // 省略关闭资源
}

6.3 confirm、Return机制在SpringBoot中的应用。

1.修改配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
spring:
  rabbitmq:
    host: 10.15.0.9
    port: 5672
    username: ems
    password: 123
    virtual-host: /ems
    # simple机制
    publisher-confirm-type: simple
    # Return机制
    publisher-returns: true
    listener:  
      simple: 
        acknowledge-mode: manual # 支持手动ACK

2.配置类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
/**
 * Confirm 与 Return的配置类
 * 需要实现ConfirmCallback与ReturnCallback接口。并重写对应的方法。
 * 需要让RabbitMQ知道这里开启了Confirm 和 Return。
 */
@Component
public class PublisherConfirmAndReturnConfig implements 
    						RabbitTemplate.ConfirmCallback, 
							RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct  // init-method
    public void initMethod() {
        // 需要让RabbitMQ知道这里开启了Confirm 和 Return。
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    // 基于Confirm的回调方法
    @Override
    public void confirm(CorrelationData correlationData, 
                        boolean ack, 
                        String cause) {
        
        if(ack == true) {
            System.out.println("消息已经送达到Exchange");
        } else {
            System.out.println("消息未送到Exchange");
        }
    }


    // 基于Return的回调方法（消息未送到队列中，会执行此方法）
    @Override
    public void returnedMessage(Message message, 
                                int replyCode, 
                                String replyText, 
                                String exchange, 
                                String routingKey) {
        
        System.out.println("消息未送达到Queue");
    }
}

3. 生产者消费者还是和原来一样。不用改动

6.4 避免重复消费消息

重复消费消息：消费者没有给RabbitMQ一个ACK。

消费者C1从RabbitMQ拿到消息msg1，当他消费成功后，会向RabbitMQ返回ACK。但是由于某种原因，消费者C1未返回ACK。RabbitMQ会认为C1未成功消费msg1，就会把此msg1发给C2。导致msg1被执行了2次。

重复消费带来的问题：

  1. 业务是幂等性操作：重复消费消息，对此业务没有任何影响。（幂等性操作：如删除和修改，执行一次和执行n次是没有区别的）
  2. 业务是非幂等性操作：要保证消息不会被重复消费。（非幂等性操作：如添加数据，且主键还是自增的）

image-20200314221002008

解决方法：引入Redis

消费者C1从RabbitMQ拿到消息msg1，先把msg1的id放到Redis里面（id作为key，值为状态码），然后再执行其业务。

  • id - 0 （正在执行业务）
  • id - 1 （业务执行成功）

当C1消费成功后，未向RabbitMQ返回ACK。RabbitMQ会把消息msg1交给C2。C2从RabbitMQ拿到消息msg1，把msg1的id放到Redis里面时，发现此id已经存在了。

然后根据id的value可以得到以下两种情况：

  • 值为0：这个消息msg1正在被消费中。
  • 值为1：这个消息msg1已经被消费完了。

C2根据id所对应的状态值，决定是否消费msg1（0：C2什么都不做，1：直接向RabbitMQ发ACK）。

上述解决方法还存在一个问题：C1收到msg1并且执行过程中，C1出现死锁，导致C1一直处于执行业务状态，不成功也不失败，id一直存在Redis里面。解决方法，给存到Redis的id加上生存时间，如10ms。

1.生产者：

1
2
3
4
5
6
7
8
9
10
11
channel.queueDeclare("hello", true, false, false, null);

AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder()
                .deliveryMode(1)  //消息是否持久化到队列，1需要、2不需要
                .messageId(UUID.randomUUID().toString()) // 设置消息的id
                .build();

String msg = "Hello World";

channel.basicPublish("", "hello", properties, msg.getBytes());

2.消费者：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
channel.queueDeclare("hello", true, false, false, null);

channel.basicConsume("hello", true, new DefaultConsumer(channel){
  @Override
  public void handleDelivery(String consumerTag, 
                             Envelope envelope, 
                             AMQP.BasicProperties properties, 
                             byte[] body) throws IOException {
      
      // 1.获得messageId
      String messageId = properties.getMessageId();
      
      // 2.消费消息前：setnx到Redis中，指定value默认为0，存活时间10s。
      ValueOperations value = redisTemplate.opsForValue();
      String result = value.set(messageId, "0", "NX", "EX", 10);
      
      // 如果key设置成功，执行业务
      if(result != null && result.equalsIgnoreCase("OK")){
          
          // 消费消息的业务
          System.out.println("消费者1: " + new String(body));
      
          // 3.消费消息完成后：把Redis中messageId改为1。并设置ACK
          value.set(messageId, "1");
          channel.basicAck(envelope.getDeliveryTag(), false);
          
      } else {
          /*
             4.如果2中setnx失败，说明要么有人正在消费，要么消费完了。
             获取此messageId对应的value。
                如果为0，什么都不做直接return。
                如果为1，设置ACK。
          */
          String flag = value.get(messageId);
          if("1".equals(flag)){
              channel.basicAck(envelope.getDeliveryTag(), false);
          }
      }
  }
});

6.4.1 用SpringBoot实现

1.生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void contextLoads() throws Exception {
    // 设置messageId
    CorrelationData messageID = 
               new CorrelationData(UUID.randomUUID().toString());
    
    // 发消息
    rabbitTemplate.convertAndSend("exchange", "routingKey", 
                                  "HelloWorld", messageID);
}

2.消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@Autowired
RedisTemplate redisTemplate;

@RabbitListener(queues = "boot-queue")
public void getMessage(String msg, 
                       Channel channel, 
                       Message message) throws Exception {

    // 1.获得messageId
    String messageId = message.getMessageProperties().getMessageId();

    // 2.消费消息前：setnx到Redis中，指定value默认为0，存活时间10s。
    boolean result = redisTemplate
                  .opsForValue()
                  .setIfAbsent(messageId, "0", 10, TimeUnit.SECONDS);

    // 如果key设置成功，执行业务
    if (result == true) {
        // 3.消费消息：消费消息的业务
        System.out.println("收到消息" + msg);

        // 4.消费消息成功后：把Redis中messageId改为1。并设置ACK
        redisTemplate.opsForValue()
                     .set(messageId, "1", 10, TimeUnit.SECONDS);
        // 手动设置ACK
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } else {
        /*
           5.如果2中setnx失败，说明要么有人正在消费，要么消费完了。
           获取此messageId对应的value。
              如果为0，什么都不做直接return。
              如果为1，设置ACK。
        */
        String value = (String) redisTemplate.opsForValue().get(messageId);
        if ("1".equals(value)){
            // 手动设置ACK
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}

7. MQ的应用场景

7.1 异步处理

场景说明：用户注册后，需要发注册邮件和注册短信，传统的做法有两种 1.串行的方式 2.并行的方式。

  • 串行方式： 将注册信息写入数据库后，发送注册邮件，再发送注册短信，以上三个任务全部完成后才返回给客户端。 这有一个问题是：邮件、短信并不是必须的，它只是一个通知。而这种做法让客户端等待没有必要等待的东西。
    这里写图片描述

  • 并行方式：将注册信息写入数据库后，同时发送邮件和短信，以上三个任务完成后，返回给客户端。并行的方式能提高处理的时间。
    这里写图片描述

  • 消息队列：假设三个业务节点分别使用50ms，串行方式使用时间150ms，并行使用时间100ms。虽然并行已经提高的处理时间。但是，前面说过，邮件和短信对我正常的使用网站没有任何影响，客户端没有必要等着其发送完成才显示注册成功，应该是写入数据库后就返回。 消息队列: 引入消息队列后，把发送邮件,短信不是必须的业务逻辑异步处理 。
    img
    由此可以看出，引入消息队列后，用户的响应时间就等于写入数据库的时间+写入消息队列的时间(可以忽略不计)。引入消息队列后处理后，响应时间是串行的3倍,是并行的2倍。

7.2 应用解耦

场景：双11是购物狂节，用户下单后，订单系统需要通知库存系统，传统的做法就是订单系统调用库存系统的接口。

这里写图片描述

这种做法有一个缺点：当库存系统出现故障时，订单就会失败。 订单系统和库存系统高耦合。

引入消息队列：

这里写图片描述

  • 订单系统:用户下单后,订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功。

  • 库存系统:订阅下单的消息,获取下单消息，进行库操作。 就算库存系统出现故障,消息队列也能保证消息的可靠投递，不会导致消息丢失。

7.3 流量削峰

场景: 秒杀活动，一般会因为流量过大，导致应用挂掉。为了解决这个问题，一般在应用前端加入消息队列。

作用:

  1. 可以控制活动人数，超过此一定阀值的订单直接丢弃(我为什么秒杀一次都没有成功过呢^^)

  2. 可以缓解短时间的高流量压垮应用(应用程序按自己的最大处理能力获取订单)

    这里写图片描述

1.用户的请求，服务器收到之后，首先写入消息队列。加入消息队列长度超过最大值，则直接抛弃用户请求或跳转到错误页面。
2.秒杀业务根据消息队列中的请求信息，再做后续处理。

7.4举例

当系统A添加一个用户完成后，存到消息队列。然后系统B从消息队列里面拿。

7.4.1 配置文件（系统A、系统B共有）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# RabbitMQ部分的配置
spring:
  rabbitmq:
    host: 10.15.0.9
    port: 5672
    username: ems
    password: 123
    virtual-host: /ems
    # simple机制
    publisher-confirm-type: simple
    # Return机制
    publisher-returns: true
    listener:  
      simple: 
        acknowledge-mode: manual # 支持手动ACK

7.4.2 配置类（系统A、系统B共有）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Configuration
public class RabbitMQConfig {

    /**
     * 交换机名：openapi-customer-exchange
     * 是否持久化：true
     * 是否能自动删除：false
     * @return
     */
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("openapi-customer-exchange", true, false);
    }

    /**
     * 队列名：openapi-customer-queue
     */
    @Bean
    public Queue queue() {
        return new Queue("openapi-customer-queue");
    }

    /**
     * 将交换机与队列绑定
     */
    public Binding binding(Queue queue, TopicExchange topicExchange) {
        return BindingBuilder
                .bind(queue)       // 队列
                .to(topicExchange) // 交换机
                .with("openapi-customer.*"); // 路由key
    }
}

7.4.3 Service层（系统A）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Service
public class UserServiceImpl implements UserService {

    @Autowired
    private UserMapper userMapper;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    @Transactional
    public void addUser(User user) {
        // 添加用户到数据库
        userMapper.addUser(user);

        // 发送消息
        rabbitTemplate.convertAndSend(
                "openapi-customer-exchange",  // 交换机名
                "openapi-customer.add",      // 定义当前路由key
                JSON.toJSON(user));          // 要发送的数据
    }
}

7.4.3 消费者模块（系统B）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@Component
public class UserListener {

    @Autowired
    private UserService userService;

    /**
     * 根据获取到的RoutingKey，判断消息对应的业务，并做相应操作
     * @param msg：需要接收的信息 User实体类的JSON串
     * @param channel：通道
     * @param message：消息对象
     */
    @RabbitListener(queues = "openapi-customer-queue")
    public void Consume(String msg, 
                        Channel channel, 
                        Message message) throws IOException{

        // 将收到的JSON串转为相应的对象(此工具类要自己手动写)
        User user = JSON.parseJSON(msg, User.class);

        // 1.获取RoutingKey
        String routingKey = message
                          .getMessageProperties()
                          .getReceivedRoutingKey();

        // 2.使用Switch判断
        switch (routingKey){
            case "openapi-customer.add":
                // 3.做出相应操作（调用Service完成）
                userService.addUser(user);
                // 4.添加成功后，手动ACK
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }

    }
}

8. RabbitMQ的集群

8.1 集群架构

8.1.1 普通集群(副本集群)

RabbitMQ代理操作所需的所有数据/状态都在所有节点之间复制。消息队列是一个例外，消息队列默认情况下位于一个节点上，尽管它们在所有节点上都是可见且可访问的。在集群(cluster)中的节点之间复制队列。

即主节点中所有数据都可以复制到从节点。但是主节点的消息队列无法复制到从节点，虽然从节点能看到主节点消息队列上的数据。（要实现高可用，看后面的镜像集群）

  1. 架构图

image-20200320094147471

​ 核心解决问题: 当集群中某一时刻master节点宕机，可以对Quene中信息，进行备份。

  1. 集群搭建
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
# 0.集群规划
主节点mq1: 10.15.0.3  
从节点mq2: 10.15.0.4  
从节点mq3: 10.15.0.5  

# 1.克隆三台机器 主机名和ip映射
    分别改3台机器的ip地址，以mq1为例：将ip修改为10.15.0.3
vim /etc/hosts

    分别改3台机器的主机名，以mq1为例：将主机名修改为mq1
vim /etc/hostname

    改完之后重启机器。

    然后分别修改3台机器的hosts文件，做另外2台主机的 ip与主机名的映射。

# 2.分别在3个机器上安装rabbitmq（安装步骤见第一节），并同步erlang.cookie文件。
    在node1上执行：把mq1的erlang.cookie复制给mq2和mq3。
    拷贝之前先关闭3台机器的RabbitMQ服务。
scp /var/lib/rabbitmq/.erlang.cookie root@mq2:/var/lib/rabbitmq/
scp /var/lib/rabbitmq/.erlang.cookie root@mq3:/var/lib/rabbitmq/

# 3.查看3个机器的erlang.cookie是否一致：以mq1为例
cat /var/lib/rabbitmq/.erlang.cookie

# 4.后台启动rabbitmq所有节点执行如下命令，启动成功访问管理界面:
rabbitmq-server -detached 

# 5.分别在node2和node3机器上执行加入集群命令（mq1不做任何操作）:
    被加入的节点就会成为主节点。
1.停止从节点   rabbitmqctl stop_app
2.加入集群    rabbitmqctl join_cluster rabbit@mq1
3.启动服务    rabbitmqctl start_app

# 6.查看集群状态，任意节点执行:
rabbitmqctl cluster_status

# 7.如果出现如下显示,集群搭建成功:
Cluster status of node rabbit@mq3 ...
[{nodes,[{disc,[rabbit@mq1,rabbit@mq2,rabbit@mq3]}]},
{running_nodes,[rabbit@mq1,rabbit@mq2,rabbit@mq3]},
{cluster_name,<<"rabbit@mq1">>},
{partitions,[]},
{alarms,[{rabbit@mq1,[]},{rabbit@mq2,[]},{rabbit@mq3,[]}]}]

8.登录管理界面,展示如下状态:

image-20200320095613586

9.测试集群在node1上，创建队列

image-20200320095743935

10.查看node2和node3节点：都有刚刚主节点创建的队列

image-20200320095827688
image-20200320095843370

11.关闭node1节点,执行如下命令,查看node2和node3:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
rabbitmqctl stop_app
```  
![image-20200320100000347](/images/RabbitMQ/image-20200320100000347.png)  
![image-20200320100010968](/images/RabbitMQ/image-20200320100010968.png)


普通集群在生产环境中用的不多，主要有以下缺点：
1. 主节点宕机会导致整个系统不可用。
2. 主节点中未持久化的消息不会同步到从节点。
3. 不能做故障转移。


#### 8.1.2 镜像集群  
> 默认情况下，RabbitMQ集群中队列的内容位于主节点上。 这与交换和绑定相反，交换和绑定始终可以被视为在所有节点上。 可以选择跨多个节点对队列进行“镜像”。
>
> `镜像队列机制就是将队列在三个节点之间设置主从关系，消息会在三个节点之间进行自动同步，且如果其中一个节点不可用，并不会导致消息丢失或服务不可用的情况，提升MQ集群的整体高可用性。`

1. ##### 集群架构图
![image-20200320113423235](/images/RabbitMQ/image-20200320113423235.png)  

2. ##### 配置集群架构  
```bash
# 0.策略说明
rabbitmqctl set_policy [-p <vhost>] [--priority <priority>] [--apply-to <apply-to>] <name> <pattern>  <definition>
-p Vhost： 可选参数，针对指定vhost下的queue进行设置
Name:     policy的名称
Pattern: queue的匹配模式(正则表达式)
Definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode
                ha-mode:指明镜像队列的模式，有效值为 all/exactly/nodes
                        all：表示在集群中所有的节点上进行镜像
                        exactly：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定
                        nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定
                ha-params：ha-mode模式需要用到的参数
                ha-sync-mode：进行队列中消息的同步方式，有效值为automatic和manual
                priority：可选参数，policy的优先级
                
                
# 1.查看当前策略
rabbitmqctl list_policies


# 2.添加策略
rabbitmqctl set_policy ha-all '^hello' '{"ha-mode":"all","ha-sync-mode":"automatic"}' 
说明:策略正则表达式为 “^” 表示所有匹配所有队列名称  ^hello:匹配hello开头队列

# 3.删除策略
rabbitmqctl clear_policy ha-all

# 4.测试集群
赏

失败是成功之母

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

JAVA<br><br>学习永不止步<br>简简单单