执行器服务RabbitMQ中只有一个线程同时运行

Question

我已经创建了一个与具有 20 个核心的指定线程池的连接。

        ConnectionFactory factory = new ConnectionFactory();
        ....
        //specified es
        ExecutorService consumerExecutor = Executors.newFixedThreadPool(threadNum, threadFactory);
        con = factory.newConnection(consumerExecutor, addresses);

然后从此连接创建一个通道：

        final Channel channel = connection.createChannel();

并使用它来创建一个 DefaultConsumer。

虽然我发现虽然线程可以用来消费消息，但始终只有一个线程在消费消息，即使消息在服务器中大量积累。

我查看源代码并找到：

private final class WorkPoolRunnable implements Runnable {

    @Override
    public void run() {
        int size = MAX_RUNNABLE_BLOCK_SIZE;
        List<Runnable> block = new ArrayList<Runnable>(size);
        try {
            Channel key = ConsumerWorkService.this.workPool.nextWorkBlock(block, size);
            if (key == null) return; // nothing ready to run
            try {
                for (Runnable runnable : block) {
                    runnable.run();
                }
            } finally {
                if (ConsumerWorkService.this.workPool.finishWorkBlock(key)) {
                    ConsumerWorkService.this.executor.execute(new WorkPoolRunnable());
                }
            }
        } catch (RuntimeException e) {
            Thread.currentThread().interrupt();
        }
    }
}


/* Basic work selector and state transition step */
private K readyToInProgress() {
    K key = this.ready.poll();
    if (key != null) {
        this.inProgress.add(key);
    }
    return key;
}


/**
 * Return the next <i>ready</i> client,
 * and transfer a collection of that client's items to process.
 * Mark client <i>in progress</i>.
 * If there is no <i>ready</i> client, return <code><b>null</b></code>.
 * @param to collection object in which to transfer items
 * @param size max number of items to transfer
 * @return key of client to whom items belong, or <code><b>null</b></code> if there is none.
 */
public K nextWorkBlock(Collection<W> to, int size) {
    synchronized (this) {
        K nextKey = readyToInProgress();
        if (nextKey != null) {
            VariableLinkedBlockingQueue<W> queue = this.pool.get(nextKey);
            drainTo(queue, to, size);
        }
        return nextKey;
    }
}

诀窍应该在ConsumerWorkService.this.workPool.nextWorkBlock，它从就绪队列中轮询通道，并在运行回调run() 后添加到完成块中的读取队列。如果我错了，请纠正我。

这很令人困惑，因为消费者绑定到一个通道，并且在最后一个任务块完成之前，通道不会释放到队列中，这意味着线程池始终只为该消费者提供一个线程。

问题：

1. 为什么 RabbitMQ 设计这个模型
2. 我们如何优化这个问题
3. 是否可以将任务提交到handleDelivery中的独立线程池以消费消息以及确认（确保仅在任务完成后确认消息）

Answer 1

> 1. 为什么 RabbitMQ 设计这个模型

我想知道自己的原因。但是这个事实在他们的documentation:

每个 Channel 都有自己的调度线程。对于最常见的用例 每个渠道一个消费者，这意味着消费者不会阻止其他消费者 消费者。如果每个频道有多个消费者，请注意 长时间运行的消费者可能会阻止向其他人发送回调 该频道上的消费者。

> 2. 我们如何优化这个问题

您可以拥有多个通道，也可以通过将实际工作提交到另一个线程池来将消息消耗与处理分离。您可以在this article找到更多详细信息。

> 3. 把任务提交到handleDelivery中的独立线程池来消费消息和ack好不好（保证任务完成后才确认消息）

引用自docs：

使用手动确认时，重要的是要考虑 什么线程进行确认。如果它不同于 接收交付的线程（例如 Consumer#handleDelivery 将交付处理委托给不同的线程），确认 将多个参数设置为 true 是不安全的，将导致 双重确认，因此是通道级协议 关闭通道的异常。确认单个消息 时间可以是安全的。