美文网首页
解决MQ消费者消息堆积问题(MQ的消费者消息丢弃功能)

解决MQ消费者消息堆积问题(MQ的消费者消息丢弃功能)

作者: Knight_9 | 来源:发表于2020-08-17 22:42 被阅读0次

问题介绍

在开发中,遇到了这样一个问题,我们使用ActivateMQ来接收处理消息,然后调用人工智能的算法去处理数据,但是算法处理的速度太慢,跟不上消息的接收速度,限制于硬件的问题,算法也没办法增加更多的服务器来进行并发处理。所以导致消息堆积,处理的延迟越来越大,新的数据得不到处理。而我们待处理的消息,可以容忍丢弃一部分,所以想到了,如果处理不完的数据,可以丢弃,保证最新的数据能够得到处理。

生产者消费者模式介绍

生产者消费者模式,是一种常见的设计模式。该模式将消息的生成者和消费者分开,还有一个缓冲区处于生产者和消费者之间,作为一个中介。生产者把数据放入缓冲区,而消费者从缓冲区取出数据。
使用mq队列,也是一种服务之间的生产者消费者模式,其缓冲区即mq队列。而在java单应用中,生产者消费者模式,一般使用阻塞队列来实现,即BlockingQueue接口,来充当缓冲区。

BlockingQueue简介

阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作是:在队列为空时,获取元素的线程会等待队列变为非空。当队列满时,存储元素的线程会等待队列可用。生产者是往队列里添加元素的线程,消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器,而消费者也只从容器里拿元素。

BlockingQueue的接口方法:

方法\处理方式 抛出异常 返回特殊值 一直阻塞 超时退出
插入方法 add(e) offer(e) put(e) offer(e,time,unit)
移除方法 remove() poll() take() poll(time,unit)
检查方法 element() peek() 不可用 不可用

上述这些方法不再详细介绍了,大家可以查阅其他资料。

容忍丢弃的消费者实现(使用BlockingQueue实现)

消费者,使用生产者消费者模式实现。消费时,不是直接进行处理,而是将消息添加到阻塞队列中,如果指定的时间内,没有添加进去,就跳过该条消息,这条消息可丢弃,或者保存数据库等,以便后续处理。
将消息添加到阻塞队列中时,通过指定的时间,可以计算出该消费者的最低并发,如指定1s,则最低并发为1,即如果处理消息时,处理速度大于1个/s,则该消费者并发降到最低,变为1。所以为了保证mq中消息不堆积,可以设置添加阻塞队列的超时时间,和mq中接收消息的间隔一致。
如果有多个mq的监听消费者,则根据多消费者来计算,确定阻塞队列的超时时间:
阻塞队列的超时时间 = mq中接收消息的间隔 * mq监听者数量
(当然系统运行时,消息推送处理等会有耗时,阻塞队列的超时时间应当适当小一些,保证mq消息不堆积)

阻塞队列消费时,使用线程池,启动多个线程,读取阻塞队列,处理消息。

代码示例:

package com.pu.WebMagicPro;

import java.util.concurrent.*;

public class ConsumerTest {

    /**
     * 使用多线程处理消息
     */
    private Executor executor = new ThreadPoolExecutor(3, 5,
            10, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(512), // 使用有界队列,避免OOM
            new ThreadPoolExecutor.DiscardPolicy());

    /**
     * 队列长度
     */
    private static final int QUEUE_SIZE = 3;

    /**
     * 消费者处理队列
     */
    private BlockingQueue<String> queue = new ArrayBlockingQueue<>(QUEUE_SIZE);

    /**
     * 启动读取队列内容的标记
     */
    private volatile boolean readQueue = false;


    /**
     * 消费方法
     * @param content 消息内容
     */
    public void consumer(String content) {
        try {
            // 消费消息,1s未插入队列,则消费不成功
            boolean success = queue.offer(content, 1, TimeUnit.SECONDS);
            if (!success) {
                System.out.println("消息丢弃:" + content);     // 消息消费不成功,可以直接丢弃,或者保存到数据库中等,使用其他办法处理
            }

            if (!readQueue) {
                readQueue = true;
                // 启动多个线程消费队列
                for (int i = 0; i < QUEUE_SIZE; i++) {
                    executor.execute(this::startRead);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void startRead() {
        while (readQueue) {
            String content = null;
            try {
                // 读取消息
                content = queue.poll(5, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (content != null) {
                // 处理消息
                dealContent(content);
            }
        }
    }

    /**
     * 模拟实际的处理方法
     * @param content 消息
     */
    private void dealContent(String content) {
        // 模拟处理过程
        System.out.println("处理: " + content);
        try {
            // 线程睡眠5秒,模拟处理用时
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("处理完成: " + content);
    }

    public static void main(String[] args) {
        ConsumerTest consumer = new ConsumerTest();
        int size = 5;
        // 开启5个线程进行消息的推送,每个线程推送10次消息
        for (int i = 0; i < size; i++) {
            final int t = i;
            Thread thread = new Thread(() -> {
                for (int j = 0; j < 10; j++) {
                    consumer.consumer("线程:" + t + ",消息" + j);
                }
            });
            thread.start();
        }
    }
}

相关文章

网友评论

      本文标题:解决MQ消费者消息堆积问题(MQ的消费者消息丢弃功能)

      本文链接:https://www.haomeiwen.com/subject/uekjrktx.html