消息队列(MQ)

Posted xinruyi

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了消息队列(MQ)相关的知识,希望对你有一定的参考价值。

什么是消息队列

消息队列,即MQ,Message Queue。
技术图片

消息队列是典型的:生产者、消费者模型。生产者不断向消息队列中生产消息,消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的,而且只关心消息的发送和接收,没有业务逻辑的侵入,这样就实现了生产者和消费者的解耦。

结合前面所说的问题:

  • 商品服务对商品增删改以后,无需去操作索引库或静态页面,只是发送一条消息,也不关心消息被谁接收。

  • 搜索服务和静态页面服务接收消息,分别去处理索引库和静态页面。

如果以后有其它系统也依赖商品服务的数据,同样监听消息即可,商品服务无需任何代码修改。

 

AMQP和JMS

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

技术图片

 

技术图片

两者间的区别和联系:

  • JMS是定义了统一的接口,来对消息操作进行统一;AMQP是通过规定协议来统一数据交互的格式

  • JMS限定了必须使用Java语言;AMQP只是协议,不规定实现方式,因此是跨语言的。

  • JMS规定了两种消息模型;而AMQP的消息模型更加丰富

常见MQ产品

技术图片

 

 

  • ActiveMQ:基于JMS, Apache

  • RabbitMQ:基于AMQP协议,erlang语言开发,稳定性好

  • RocketMQ:基于JMS,阿里巴巴产品,目前交由Apache基金会

  • Kafka:分布式消息系统,高吞吐量

RabbitMQ

RabbitMQ是基于AMQP的一款消息管理系统

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

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

技术图片

 

技术图片

RabbitMQ基于Erlang语言开发:

技术图片

下载和安装

下载

官网下载地址:http://www.rabbitmq.com/download.html

技术图片

 

 

我们使用的是:3.4.1版本

安装

技术图片

 


五种消息模型

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

但是其实3、4、5这三种都属于订阅模型,只不过进行路由的方式不同。

技术图片

 

 

导入demo工程

我们通过一个demo工程来了解下RabbitMQ的工作方式:

技术图片

 

 

基本消息模型

说明

官方文档说明:

RabbitMQ是一个消息的代理者(Message Broker):它接收消息并且传递消息。

你可以认为它是一个邮局:当你投递邮件到一个邮箱,你很肯定邮递员会终究会将邮件递交给你的收件人。与此类似,RabbitMQ 可以是一个邮箱、邮局、同时还有邮递员。

不同之处在于:RabbitMQ不是传递纸质邮件,而是二进制的数据

基本消息模型图:

技术图片

 

在上图的模型中,有以下概念:

  • P:生产者,也就是要发送消息的程序

  • C:消费者:消息的接受者,会一直等待消息到来。

  • queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。

生产者

连接工具类:

public class ConnectionUtil 
    /**
     * 建立与RabbitMQ的连接
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception 
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("192.168.56.101");
        //端口
        factory.setPort(5672);
        //设置账号信息,用户名、密码、vhost
        factory.setVirtualHost("/leyou");
        factory.setUsername("leyou");
        factory.setPassword("leyou");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    

  

生产者发送消息:

public class Send 

    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 从连接中创建通道,使用通道才能完成消息相关的操作
        Channel channel = connection.createChannel();
        // 声明(创建)队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 消息内容
        String message = "Hello World!";
        // 向指定的队列中发送消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        
        System.out.println(" [x] Sent ‘" + message + "‘");

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

  

控制台:

技术图片

 

web控制台查看消息

进入队列页面,可以看到新建了一个队列:simple_queue

技术图片

 

 

点击队列名称,进入详情页,可以查看消息:

技术图片

 

在控制台查看消息并不会将消息消费,所以消息还在。

消费者获取消息

public class Recv 
    private final static String QUEUE_NAME = "simple_queue";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        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 + "!");
            
        ;
        // 监听队列,第二个参数:是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    

  

控制台:

技术图片

 

这个时候,队列中的消息就没了:

技术图片

 

消费者的消息确认机制

通过刚才的案例可以看出,消息一旦被消费者接收,队列中的消息就会被删除。

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

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

  • 自动ACK:消息一旦被接收,消费者自动发送ACK

  • 手动ACK:消息接收后,不会发送ACK,需要手动调用

大家觉得哪种更好呢?

这需要看消息的重要性:

  • 如果消息不太重要,丢失也没有影响,那么自动ACK会比较方便

  • 如果消息非常重要,不容丢失。那么最好在消费完成后手动ACK,否则接收消息后就自动ACK,RabbitMQ就会把消息从队列中删除。如果此时消费者宕机,那么消息就丢失了。

我们之前的测试都是自动ACK的,如果要手动ACK,需要改动我们的代码:

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);
    

  

注意到最后一行代码:

channel.basicConsume(QUEUE_NAME, false, consumer);
 

如果第二个参数为true,则会自动进行ACK;如果为false,则需要手动ACK。方法的声明:

技术图片

 

 

work消息模型

说明

在刚才的基本模型中,一个生产者,一个消费者,生产的消息直接被消费者消费。比较简单。

Work queues,也被称为(Task queues),任务模型。

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

技术图片

 

角色:

  • P:生产者:任务的发布者

  • C1:消费者,领取任务并且完成任务,假设完成速度较慢

  • C2:消费者2:领取任务并完成任务,假设完成速度快

生产者

生产者与案例1中的几乎一样:

public class Send 
    private final static String QUEUE_NAME = "test_work_queue";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 循环发布任务
        for (int i = 0; i < 50; i++) 
            // 消息内容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent ‘" + message + "‘");

            Thread.sleep(i * 2);
        
        // 关闭通道和连接
        channel.close();
        connection.close();
    

  

不过这里我们是循环发送50条消息。

消费者1

技术图片

 

消费者2

技术图片

 

 

与消费者1基本类似,就是没有设置消费耗时时间。

这里是模拟有些消费者快,有些比较慢。

 

接下来,两个消费者一同启动,然后发送50条消息:

技术图片

 

 

可以发现,两个消费者各自消费了25条消息,而且各不相同,这就实现了任务的分发。

 

能者多劳

刚才的实现有问题吗?

  • 消费者1比消费者2的效率要低,一次任务的耗时较长

  • 然而两人最终消费的消息数量是一样的

  • 消费者2大量时间处于空闲状态,消费者1一直忙碌

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

怎么实现呢?

我们可以修改设置,让消费者同一时间只接收一条消息,这样处理完成之前,就不会接收更多消息,就可以让处理快的人,接收更多消息 :

技术图片

 

 

再次测试:

技术图片

 

订阅模型分类

订阅模型示意图:

技术图片

 

前面2个案例中,只有3个角色:

  • P:生产者,也就是要发送消息的程序

  • C:消费者:消息的接受者,会一直等待消息到来。

  • queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。

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

  • P:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给X(交换机)

  • C:消费者,消息的接受者,会一直等待消息到来。

  • Queue:消息队列,接收消息、缓存消息。

  • Exchange:交换机,图中的X。一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。Exchange有以下3种类型:

    • Fanout:广播,将消息交给所有绑定到交换机的队列

    • Direct:定向,把消息交给符合指定routing key 的队列

    • Topic:通配符,把消息交给符合routing pattern(路由模式) 的队列

Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!

 

订阅模型-Fanout

Fanout,也称为广播。

流程说明

流程图:

技术图片

 

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

  • 1) 可以有多个消费者

  • 2) 每个消费者有自己的queue(队列)

  • 3) 每个队列都要绑定到Exchange(交换机)

  • 4) 生产者发送的消息,只能发送到交换机,交换机来决定要发给哪个队列,生产者无法决定。

  • 5) 交换机把消息发送给绑定过的所有队列

  • 6) 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

生产者

两个变化:

  • 1) 声明Exchange,不再声明Queue

  • 2) 发送消息到Exchange,不再发送到Queue

public class Send 

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        
        // 声明exchange,指定类型为fanout
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        
        // 消息内容
        String message = "Hello everyone";
        // 发布消息到Exchange
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [生产者] Sent ‘" + message + "‘");

        channel.close();
        connection.close();
    

  

消费者1

public class Recv 
    private final static String QUEUE_NAME = "fanout_exchange_queue_1";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 定义队列的消费者
        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(" [消费者1] received : " + msg + "!");
            
        ;
        // 监听队列,自动返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    

  

要注意代码中:队列需要和交换机绑定

消费者2

public class Recv2 
    private final static String QUEUE_NAME = "fanout_exchange_queue_2";

    private final static String EXCHANGE_NAME = "fanout_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        
        // 定义队列的消费者
        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(" [消费者2] received : " + msg + "!");
            
        ;
        // 监听队列,手动返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);
    

  

测试

我们运行两个消费者,然后发送1条消息:

技术图片

 

 

订阅模型-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 的消息

生产者

此处我们模拟商品的增删改,发送消息的RoutingKey分别是:insert、update、delete

public class Send 
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange,指定类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 消息内容
        String message = "商品新增了, id = 1001";
        // 发送消息,并且指定routing key 为:insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服务:] Sent ‘" + message + "‘");

        channel.close();
        connection.close();
    

  

消费者1

我们此处假设消费者1只接收两种类型的消息:更新商品和删除商品。

public class Recv 
    private final static String QUEUE_NAME = "direct_exchange_queue_1";
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定义队列的消费者
        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(" [消费者1] received : " + msg + "!");
            
        ;
        // 监听队列,自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    

  

消费者2

我们此处假设消费者2接收所有类型的消息:新增商品,更新商品和删除商品。

public class Recv2 
    private final static String QUEUE_NAME = "direct_exchange_queue_2";
    private final static String EXCHANGE_NAME = "direct_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机,同时指定需要订阅的routing key。订阅 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定义队列的消费者
        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(" [消费者2] received : " + msg + "!");
            
        ;
        // 监听队列,自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    

  

测试

我们分别发送增、删、改的RoutingKey,发现结果:

技术图片

 

 

订阅模型-Topic

说明

Topic类型的ExchangeDirect相比,都是可以根据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 都会被匹配

生产者

使用topic类型的Exchange,发送消息的routing key有3种: item.isnertitem.updateitem.delete

public class Send 
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange,指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        // 消息内容
        String message = "新增商品 : id = 1001";
        // 发送消息,并且指定routing key 为:insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes());
        System.out.println(" [商品服务:] Sent ‘" + message + "‘");

        channel.close();
        connection.close();
    

  

消费者1

我们此处假设消费者1只接收两种类型的消息:更新商品和删除商品

public class Recv 
    private final static String QUEUE_NAME = "topic_exchange_queue_1";
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机,同时指定需要订阅的routing key。需要 update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

        // 定义队列的消费者
        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(" [消费者1] received : " + msg + "!");
            
        ;
        // 监听队列,自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    

  

消费者2

我们此处假设消费者2接收所有类型的消息:新增商品,更新商品和删除商品。

  

/**
 * 消费者2
 */
public class Recv2 
    private final static String QUEUE_NAME = "topic_exchange_queue_2";
    private final static String EXCHANGE_NAME = "topic_exchange_test";

    public static void main(String[] argv) throws Exception 
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 绑定队列到交换机,同时指定需要订阅的routing key。订阅 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");

        // 定义队列的消费者
        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(" [消费者2] received : " + msg + "!");
            
        ;
        // 监听队列,自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    

  

 

以上是关于消息队列(MQ)的主要内容,如果未能解决你的问题,请参考以下文章

消息队列(mq)是啥?

MQ消息队列及常见MQ比较

深入消息队列MQ,看这篇就够了!

消息队列篇—常用消息队列MQ产品介绍及对比

MQ消息队列

消息队列MQ 之 Kafka