RabbitMQ工作队列之公平分发消息与消息应答(ACK)

Stella981
• 阅读 317

上篇文章中,我们讲了工作队列轮询的分发模式,该模式无论有多少个消费者,不管每个消费者处理消息的效率,都会将所有消息平均的分发给每一个消费者,也就是说,大家最后各自消费的消息数量都是一样多的。由此也就引发我们今天要介绍的公平分发模式。

消息应答(ACK)

RabbitMQ工作队列之公平分发消息与消息应答(ACK)

消息丢失

我们之前的所有代码,如果消息队列将消息分发给消费者,那么就会从队列中删除,如果在我们处理任务的过程中,处理失败或者服务器宕机,那么这条消息肯定得不到执行,就会出现丢失。

我们所设想的如果任务在处理的过程中,如果服务器宕机等原因造成消息未被正常消费,那么必须分发给其他的消费者再次进行消费,这样及时服务器宕机也不会丢失任何的消息了。

ACK

所以ACK,就是消息应答机制,我们之前写的代码都是开启了自动应答,所以如果我们的消息没被正常消费,就会丢失。

要想确保消息不丢失,就必须将ACK自动应答关闭掉,在我们处理消息的流程中,如果消息正常被处理,那么最后进行手动应答,告诉队列我们正常消费了消息。

超时

RabbitMQ它是没有我们平常所见到的超时时间限制的,只要当消费者服务宕机,消息才会被重新分发,哪怕处理这条消息需要花费很长的时间。

公平分发模式

缺陷

我们提供多个消费者,目的就是为了提高系统的性能,提升系统处理任务的速度,如果将消息平均的分发给每个消费者,那么处理消息快的服务是不是会空闲下来,而处理慢的服务可能会阻塞等待处理,这样的场景是我们不愿意看到的。所以有了今天要说的分发模式,公平分发

能者多劳

所谓的公平分发,其实用能者多劳描述更为贴切,根据名字就可以知道,谁有能力处理更多的任务,那么就交给谁处理,防止消息的挤压。

那么想要实现公平分发,那么必须要将自动应答改为手动应答。这是公平分发的前提。

代理

消息生产者

public class Send {

    public static final String QUEUE_NAME = "test_word_queue";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        // 获取连接
        Connection connection = MQConnectUtil.getConnection();

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

        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        for (int i = 0; i < 10; i++) {

            String msg = "消息:" + i;

            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());

            Thread.sleep(i * 20);

            System.out.println(msg);
        }

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

消费者1

我们在消费者中设置了channel.basicQos(1);这样一个参数,这个意思就是表示,此消费者每次最多只接收一条消息进行处理,只有将消息处理结束,手动应答之后,下一条消息才会被分发进来。

public class Consumer1 {

    public static final String QUEUE_NAME = "test_word_queue";

    public static void main(String[] args) throws Exception {

        // 获取连接
        Connection connection = MQConnectUtil.getConnection();

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

        // 一次仅接受一条未经确认的消息
        channel.basicQos(1);

        // 队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                String msg = new String(body, StandardCharsets.UTF_8);

                System.out.println("消费者[1]-内容:" + msg);

                Thread.sleep(2 * 1000);

                // 手动回执消息
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };

        // 监听队列,将自动应答方式改为false,关闭自动应答机制
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

消费者2

public class Consumer2 {

    public static final String QUEUE_NAME = "test_word_queue";

    public static void main(String[] args) throws Exception {

        // 获取连接
        Connection connection = MQConnectUtil.getConnection();

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

        // 队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        channel.basicQos(1);

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                String msg = new String(body, StandardCharsets.UTF_8);

                System.out.println("消费者[2]-内容:" + msg);

                Thread.sleep(1000);

                // 手动回执消息
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };

        // 监听队列,需要将自动应答方式改为false
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

消费结果

那么结果就会像我们之前预想的那样,由于消费者2消费消息花费的时间比消费者1更少,所以消费者2处理的消息的数量要比消费者1处理的消息的数量要多。这里我就不贴图了,大家可以敲代码进行尝试。


今天的文章到这里就结束了,下篇呢,会给介绍介绍另外一种模式,发布订阅模式

日拱一卒,功不唐捐

更多内容请关注:

RabbitMQ工作队列之公平分发消息与消息应答(ACK)

点赞
收藏
评论区
推荐文章
blmius blmius
2年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
Stella981 Stella981
2年前
RabbitMQ如何高效的消费消息
在上篇介绍了如何简单的发送一个消息队列之后,我们本篇来看下RabbitMQ的另外一种模式,工作队列。什么是工作队列我们上篇文章说的是,一个生产者生产了消息被一个消费者消费了,如下图!(https://usergoldcdn.xitu.io/2020/5/15/1721768c1b303014?w1824&h55
Stella981 Stella981
2年前
Spring Boot(七):RabbitMQ 详解
一、RabbitMQ简介RabbitMQ即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用。消息中间件在互联网公司的使用中越来越多,消息中间件最主要的作用是解耦,中间件最标准的用法是生产者生产消息传送到队列,消费者从队列中拿取消息并处理,生产者不用关心是谁来消费,消费者不用关心谁在生产消息,从而达到解耦的
Wesley13 Wesley13
2年前
3.rabbitmq
rabbitmq发布订阅模式模型组成一个消费者Producer,一个交换机Exchange,多个消息队列Queue,多个消费者Consumer一个生产者,多个消费者,每一个消费者都有自己的一个队列,生产者没有将消息直接发送到队列,而是发送到了交换机,每个队列绑定交换机,生产者发送
Easter79 Easter79
2年前
SpringMVC中配置RabbitMQ
        RabbitMQ是工作在amqp协议(advancedmessagequeueprotocal,高级消息队列协议)上的一个消息中间件。它通过一个生产者消费者模型来处理应用中产生的消息。        除了生产者和消费者,此模型中另外一个重要的概念叫“工作队列”,也称为“任务队列”(TaskQueue),任务队列背后的核心想法是避免
Wesley13 Wesley13
2年前
JMS消息的概念解释
1、默认生产者消息是持久的:会存数据库\消费者的持久:createDurableSubscriber是指消费者能收到所有它订阅时间点之后的消息,即使消费者注册后关闭,当它重启就能收到注册时间点之后所有的消息;即当此消费用户ID(AAA)在producer发送之前就已经注册,那么此id能收到producer发送的所有消息,如果是在produce
Wesley13 Wesley13
2年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
4个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这