问题介绍

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

生产者消费者模式介绍

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

BlockingQueue简介

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

BlockingQueue的接口方法:

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

容忍丢弃的消费者实现（使用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();
        }
    }
}