如何在 Java 中运行轮询阻塞消费者？答案

【问题标题】：How to run a polling blocking consumer in Java?如何在 Java 中运行轮询阻塞消费者？
【发布时间】：2014-11-30 02:01:26
【问题描述】：

我有一些服务既从入站队列消费又产生到某个出站队列（由该服务创建的另一个线程接收消息并将它们“传输”到目的地）。

目前我使用两个普通的Threads，如下面的代码所示，但我知道通常你不应该再使用它们，而是使用更高级别的抽象，比如ExecutorService。

这对我来说有意义吗？更具体地说，我的意思是->

会减少代码吗？
让代码在失败时更加健壮？
允许更平滑的线程终止？（这在运行测试时很有帮助）

我在这里遗漏了什么重要的东西吗？（也许来自 java.util.concurrent 的一些其他类）

// called on service startup
private void init() {
    // prepare everything here
    startInboundWorkerThread();
    startOutboundTransporterWorkerThread();
}

private void startInboundWorkerThread() {
    InboundWorkerThread runnable = injector.getInstance(InboundWorkerThread.class);
    inboundWorkerThread = new Thread(runnable, ownServiceIdentifier);
    inboundWorkerThread.start();
}

// this is the Runnable for the InboundWorkerThread 
// the runnable for the transporter thread looks almost the same
@Override
public void run() {       
    while (true) {
        InboundMessage message = null;
        TransactionStatus transaction = null;

        try {
            try {
                transaction = txManager.getTransaction(new DefaultTransactionDefinition());
            } catch (Exception ex) {
                // logging
                break;
            }

            // blocking consumer
            message = repository.takeOrdered(template, MESSAGE_POLL_TIMEOUT_MILLIS);
            if (message != null) {                   
                handleMessage(message);
                commitTransaction(message, transaction);
            } else {
                commitTransaction(transaction);
            }
        } catch (Exception e) {
            // logging
            rollback(transaction);
        } catch (Throwable e) {
            // logging
            rollback(transaction);
            throw e;
        }

        if (Thread.interrupted()) {
            // logging
            break;
        }
    }

    // logging
}

// called when service is shutdown
// both inbound worker thread and transporter worker thread must be terminated
private void interruptAndJoinWorkerThread(final Thread workerThread) {
    if (workerThread != null && workerThread.isAlive()) {
        workerThread.interrupt();

        try {
            workerThread.join(TimeUnit.SECONDS.toMillis(1));
        } catch (InterruptedException e) {
            // logging
        }
    }
}

【问题讨论】：

标签： java multithreading concurrency blocking java.util.concurrent

【解决方案1】：

使用 ThreadPools 对我的主要好处来自于将工作结构化为单个、独立且通常很短的作业，以及在 ThreadPools 私有 Workers 中更好地抽象线程。有时您可能希望更直接地访问它们，以了解它们是否仍在运行等。但通常有更好的、以工作为中心的方法来做到这一点。

至于处理失败，您可能希望提交自己的ThreadFactory 以使用自定义UncaughtExceptionHandler 创建线程，一般来说，您的Runnable 作业也应该提供良好的异常处理，以便记录更多信息关于失败的具体工作。使这些工作成为非阻塞的，因为您不想让被阻塞的工作人员填满您的线程池。在作业排队之前移动阻塞操作。

通常情况下，ExecutorServices 提供的shutdown 和shutdownNow 与您的作业中的适当中断处理相结合，可以顺利终止作业。

【讨论】：