关于消息队列RabbitMQ的那些事

发表于 | 更新于 | 分类于 | 评论数： | 阅读次数：
本文字数： 8.9k | 阅读时长 ≈ 16 分钟

1.消息队列（MQ）

关于消息队列那点事

1.1什么是消息队列

消息队列，即MQ，Message Queue。

1527063872737

1.2 AMQP和JMS

MQ是消息通信的模型，并不是具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。

1527064480681

正如上图所说,队列的使用除去了接受和发送应用程序同时执行的要求,消息队列是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。
1527064487042
两者间的区别和联系：

  • JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
  • JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
  • JMS规定了两种消息模型；而AMQP的消息模型更加丰富

JMS更适合于java服务之间的应用通信,而AMQP可以跨语言通信,消息模型更加丰富,在javaweb的开发中,构建springcloud应用的时用到了消息队列,springcloud传递了微服务这个概念,也许不同的服务是由不同的语言进行实现,AMQP这个消息队列协议更加适合一些.

1.3 常见MQ产品

1527064606029

  • ActiveMQ：基于JMS
  • RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
  • RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
  • Kafka：分布式消息系统，高吞吐量1527064762982

其中RabbitMQ,稳定性较好,是一个开源的,基于AMQP协议的完整的可复用的企业消息系统,支持主流的操作系统,Linux,Windows,MacOs等.支持多种开发语言
官网： http://www.rabbitmq.com/

官方教程：http://www.rabbitmq.com/getstarted.html

2.五种消息模型

2.1.基本消息模型

RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。

但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。
1527068544487
RabbitMQ是一个消息代理：它接受和转发消息。 你可以把它想象成一个邮局：当你把邮件放在邮箱里时，你可以确定邮差先生最终会把邮件发送给你的收件人。 在这个比喻中，RabbitMQ是邮政信箱，邮局和邮递员。

RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。
1527070619131
P（producer/ publisher）：生产者，一个发送消息的用户应用程序。

C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序

队列（红色区域）：rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。

总之：

生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。

我们将用Java编写两个程序;发送单个消息的生产者，以及接收消息并将其打印出来的消费者。我们将详细介绍Java API中的一些细节，这是一个消息传递的“Hello World”。

我们将调用我们的消息发布者（发送者）Send和我们的消息消费者（接收者）Recv。发布者将连接到RabbitMQ，发送一条消息，然后退出。

2.2.work消息模型

工作队列或者竞争消费者模式
在这里插入图片描述
在第一篇教程中，我们编写了一个程序，从一个命名队列中发送并接受消息。在这里，我们将创建一个工作队列，在多个工作者之间分配耗时任务。

工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多工人时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。

这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。

接下来我们来模拟这个流程：

​ P：生产者：任务的发布者

​ C1：消费者，领取任务并且完成任务，假设完成速度较快

​ C2：消费者2：领取任务并完成任务，假设完成速度慢

面试题：避免消息堆积？

1） 采用workqueue，多个消费者监听同一队列。

2）接收到消息以后，而是通过线程池，异步消费。

2.3.订阅模型分类

在这里插入图片描述
前面2个案例中，只有3个角色：

  • P：生产者，也就是要发送消息的程序
  • C：消费者：消息的接受者，会一直等待消息到来。
  • queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

而在订阅模型中，多了一个exchange角色，而且过程略有变化：

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

2.3.订阅模型-Fanout

Fanout，也称为广播。

流程说明

流程图：
在这里插入图片描述
在广播模式下，消息发送流程是这样的：

  • 1） 可以有多个消费者
  • 2） 每个消费者有自己的queue（队列）
  • 3） 每个队列都要绑定到Exchange（交换机）
  • 4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
  • 5） 交换机把消息发送给绑定过的所有队列
  • 6） 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

2.4.订阅模型-Direct

说明

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

在Direct模型下：

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

流程图：
在这里插入图片描述
图解：

  • P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
  • X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
  • C1：消费者，其所在队列指定了需要routing key 为 error 的消息
  • C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

2.5.订阅模型-Topic

说明

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

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

通配符规则：

​ #：匹配一个或多个词

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

举例：

​ audit.#：能够匹配audit.irs.corporate 或者 audit.irs

​ audit.*：只能匹配audit.irs

图示：
在这里插入图片描述
解释：

  • 红色Queue：绑定的是usa.# ，因此凡是以 usa.开头的routing key 都会被匹配到
  • 黄色Queue：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配

例子2：

1532766712166

在这个例子中，我们将发送所有描述动物的消息。消息将使用由三个字（两个点）组成的routing key发送。路由关键字中的第一个单词将描述速度，第二个颜色和第三个种类：“..”。

我们创建了三个绑定：Q1绑定了绑定键“ .orange.”，Q2绑定了“..rabbit”和“lazy.＃”。

Q1匹配所有的橙色动物。

Q2匹配关于兔子以及懒惰动物的消息。

练习，生产者发送如下消息，会进入那个队列：

quick.orange.rabbit Q1 Q2

lazy.orange.elephant Q1 Q2

quick.orange.fox Q1

lazy.pink.rabbit Q2

quick.brown.fox 不匹配任意队列，被丢弃

quick.orange.male.rabbit 不匹配任意队列，被丢弃

orange 不匹配任意队列，被丢弃

3.持久化

如何避免消息丢失？

1） 消费者的ACK机制。可以防止消费者丢失消息。

2） 但是，如果在消费者消费之前，MQ就宕机了，消息就没了。

是可以将消息进行持久化呢？

要将消息持久化，前提是：队列、Exchange都持久化

交换机持久化

在这里插入图片描述

队列持久化

在这里插入图片描述

消息持久化

1527088984029

消费者的消息确认机制(Acknowlage)

消息一旦被消费者接收，队列中的消息就会被删除。

那么问题来了：RabbitMQ怎么知道消息被接收了呢？

这就要通过消息确认机制（Acknowlege）来实现了。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：

  • 自动ACK：消息一旦被接收，消费者自动发送ACK
  • 手动ACK：消息接收后，不会发送ACK，需要手动调用

大家觉得哪种更好呢？

这需要看消息的重要性：

  • 如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便
  • 如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK，RabbitMQ就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。
    我们之前的测试都是自动ACK的，如果要手动ACK，需要改动我们的代码：
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
public class Recv2 {
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        final Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
                // 手动进行ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 监听队列，第二个参数false，手动进行ACK
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

注意到最后一行代码：

1
channel.basicConsume(QUEUE_NAME, false, consumer);

如果第二个参数为true，则会自动进行ACK；如果为false，则需要手动ACK。方法的声明：
在这里插入图片描述

能者多劳

刚才的实现有问题吗？

  • 消费者1比消费者2的效率要低，一次任务的耗时较长
  • 然而两人最终消费的消息数量是一样的
  • 消费者2大量时间处于空闲状态，消费者1一直忙碌

现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。

怎么实现呢？

我们可以修改设置，让消费者同一时间只接收一条消息，这样处理完成之前，就不会接收更多消息，就可以让处理快的人，接收更多消息 ：
在这里插入图片描述
再次测试：
在这里插入图片描述

以上原生方法语句较为复杂,日常使用为springboot整合,自动配置的代码.

4.Spring AMQP

4.1.简介

Sprin有很多不同的项目，其中就有对AMQP的支持：
在这里插入图片描述
Spring AMQP的页面：http://projects.spring.io/spring-amqp/
1527089365281
注意这里一段描述：
在这里插入图片描述
Spring-amqp是对AMQP协议的抽象实现，而spring-rabbit 是对协议的具体实现，也是目前的唯一实现。底层使用的就是RabbitMQ。

4.2.依赖和配置

添加AMQP的启动器：

1
2
3
4
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

在application.yml中添加RabbitMQ地址：

1
2
3
4
5
6
spring:
  rabbitmq:
    host: 192.168.1.102
    username: test
    password: test
    virtual-host: /test

4.3.监听者

在SpringAmqp中，对消息的消费者进行了封装和抽象，一个普通的JavaBean中的普通方法，只要通过简单的注解，就可以成为一个消费者。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Component
public class Listener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.test.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.test.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC
            ),
            key = {"#.#"}))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }
}
  • @Componet：类上的注解，注册到Spring容器
  • @RabbitListener：方法上的注解，声明这个方法是一个消费者方法，需要指定下面的属性：
    • bindings：指定绑定关系，可以有多个。值是@QueueBinding的数组。@QueueBinding包含下面属性：
      • value：这个消费者关联的队列。值是@Queue，代表一个队列
      • exchange：队列所绑定的交换机，值是@Exchange类型
      • key：队列和交换机绑定的RoutingKey

类似listen这样的方法在一个类中可以写多个，就代表多个消费者。

4.4.AmqpTemplate

Spring最擅长的事情就是封装，把他人的框架进行封装和整合。

Spring为AMQP提供了统一的消息处理模板：AmqpTemplate，非常方便的发送消息，其发送方法：

在这里插入图片描述
红框圈起来的是比较常用的3个方法，分别是：

  • 指定交换机、RoutingKey和消息体
  • 指定消息
  • 指定RoutingKey和消息，会向默认的交换机发送消息

4.5.测试代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqDemo {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSend() throws InterruptedException {
        String msg = "hello, Spring boot amqp";
        this.amqpTemplate.convertAndSend("spring.test.exchange","a.b", msg);
        // 等待10秒后再结束
        Thread.sleep(10000);
    }
}

运行后查看日志:
在这里插入图片描述
至此,Spring AMQP 成功使用.

4.6.配置文件

完善发送方application.yml配置:
我们在application.yml中添加一些有关RabbitMQ的配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
spring:
  rabbitmq:
    host: 192.168.1.103
    username: test
    password: test
    virtual-host: /test
    template:
      retry:
        enabled: true
        initial-interval: 10000ms
        max-interval: 300000ms
        multiplier: 2
      exchange: test.item.exchange
    publisher-confirms: true
  • template：有关AmqpTemplate的配置
    • retry：失败重试
      • enabled：开启失败重试
      • initial-interval：第一次重试的间隔时长
      • max-interval：最长重试间隔，超过这个间隔将不再重试
      • multiplier：下次重试间隔的倍数，此处是2即下次重试间隔是上次的2倍
    • exchange：缺省的交换机名称，此处配置后，发送消息如果不指定交换机就会使用这个
  • publisher-confirms：生产者确认机制，确保消息会正确发送，如果发送失败会有错误回执，从而触发重试

完善接受方application.yml配置:

1
2
3
4
5
6
spring:
  rabbitmq:
    host: 192.168.56.101
    username: test
    password: test
    virtual-host: /test

这里只是接收消息而不发送，所以不用配置template相关内容。

4.7 测试

查看RabbitMQ控制台

重新启动项目，并且登录RabbitMQ管理界面：http://192.168.1.103:15672

可以看到，交换机已经创建出来了,队列也已经创建完毕,并且队列都已经绑定到交换机

##

特别感谢:传智播客

-------------本文结束感谢您的阅读-------------
愿你所有幸运,都不期而遇;愿你所有美好,都如约而至。