设置读卡器、解码器和消费线程

Question

我有一个相当标准的生产者和消费者线程：

• producer将文件中的字节读取并解码到阻塞队列中
• 消费者正在轮询队列中的项目

碰巧解码过程是一个瓶颈，可能会受益于更多的CPU。这是制作人70%的时间。如果我引入“解码器”线程，我会获得任何显著的性能吗？

• 生产者从文件中读取字节到阻塞“对象”队列

由于内存不足，我需要使用一个队列——无法承受两个队列（字节/项），所以我猜对象“转换”开销会出现吗？

有没有关于如何实现这个3线程解决方案的想法？

谢谢你！

Answer 1

2个队列，其中一个用于保存多个消费者解码的未编码对象。

多个消费者将解码并将解码后的对象写入第二个队列，最终消费者将从中消费。

确保避免死锁（除非你真的知道你在做什么，否则使用通知所有（）not通知（））

Answer 2

您应该为生产者和消费者调整线程池——例如，如果消费者相对于生产者的速度太快，那么它的线程池可以分配比生产者的线程池更少的线程。这应该会导致吞吐量的显著增加。生产者和消费者线程的比例应该被调整（例如3:1）。

在类似的线路上，您可以有三个线程池，其中生产者（读者）和消费者的线程数较少，而解码器（转换器）的线程池的线程数较多。我不确定你是否需要代码示例，在这种情况下，你应该分享你目前拥有的。我会从生产者和消费者的线程池大小为1，转换器（解码器）的线程池大小为5开始，然后测量瓶颈是什么（如果吞吐量满足您的期望）

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

public class ProducerDecoderConsumer {
    /**
     * @param args
     */
    public static void main(String[] args) {
        BlockingQueue<Integer> inputQueue = new PriorityBlockingQueue<Integer>();
        BlockingQueue<String> outputQueue = new PriorityBlockingQueue<String>();
        ExecutorService reader = Executors.newSingleThreadExecutor();
        reader.submit(new Producer(inputQueue));
        ExecutorService decoder = Executors.newFixedThreadPool(5);
        decoder.submit(new Transformer(inputQueue, outputQueue));
        ExecutorService writer = Executors.newSingleThreadExecutor();
        writer.submit(new Consumer(outputQueue));

    }

    private static class Producer implements Callable<Void> {
        final BlockingQueue<Integer> queue;

        public Producer(final BlockingQueue<Integer> pQueue) {
            queue = pQueue;
        }

        @Override
        public Void call() throws Exception {
            try {
                Random random = new Random();
                while (true) {
                    queue.put(random.nextInt());
                }
            } catch (Exception e) {

            }
            return null;
        }
    }

    private static class Transformer implements Callable<Void> {
        final BlockingQueue<Integer> inputQueue;

        final BlockingQueue<String> outputQueue;

        public Transformer(final BlockingQueue<Integer> pInputQueue, final BlockingQueue<String> pOutputQueue) {
            inputQueue = pInputQueue;
            outputQueue = pOutputQueue;
        }

        @Override
        public Void call() throws Exception {
            try {
                while (true) {
                    Integer input = inputQueue.take();
                    String output = String.valueOf(input); // decode input to output
                    outputQueue.put(output); // output
                }
            } catch (Exception e) {

            }
            return null;
        }
    }

    private static class Consumer implements Callable<Void> {
        final BlockingQueue<String> queue;

        public Consumer(final BlockingQueue<String> pQueue) {
            queue = pQueue;
        }

        @Override
        public Void call() throws Exception {
            try {
                while (true) {
                    System.out.println(queue.take());
                }
            } catch (Exception e) {

            }
            return null;
        }
    }
}

我已经添加了一些代码来说明这个想法——我使用的是两个阻塞队列，不同于你问题中提到的单个队列，因为我不认为仅仅拥有额外的队列会有开销——我建议使用探查器来演示这样的事情。不过，我希望您觉得它很有用，如果您真的觉得有必要，可以将其改进为单队列模型。