Java-在注释中的单独线程上处理时忽略异常

Question

我们有一个注释，允许我们使用一个民意调查消费者来消费Kafka消息。它是为长时间运行的作业而设计的，因此一个线程正在处理消息，而另一个线程仍然可以用于轮询，以防止Kafka认为我们的服务失败，并重新平衡消费者。

我们正在使用Spring AOP。

课程：

@Aspect
@Component
@Slf4j
public class PollableStreamListenerAspect {

  private final ExecutorService executor = Executors.newFixedThreadPool(1);

  private volatile boolean paused = false;

  @Around(value = "@annotation(pollableStreamListener) && args(dataCapsule,..)")
  public void receiveMessage(ProceedingJoinPoint joinPoint,
      PollableStreamListener pollableStreamListener, Object dataCapsule) {
    if (dataCapsule instanceof Message) {
      Message<?> message = (Message<?>) dataCapsule;
      AcknowledgmentCallback callback = StaticMessageHeaderAccessor
          .getAcknowledgmentCallback(message);
      callback.noAutoAck();

      if (!paused) {
        // The separate thread is not busy with a previous message, so process this message:
        Runnable runnable = () -> {
          try {
            paused = true;

            // Call method to process this Kafka message
            joinPoint.proceed();

            callback.acknowledge(Status.ACCEPT);
          } catch (Throwable e) {
            callback.acknowledge(Status.REJECT);
            throw new PollableStreamListenerException(e);
          } finally {
            paused = false;
          }
        };

        executor.submit(runnable);
      } else {
        // The separate thread is busy with a previous message, so re-queue this message for later:
        callback.acknowledge(Status.REQUEUE);
      }
    }
  }

}

我们一次只想处理一条消息，因此暂停标志用于决定是现在处理消息，还是以后重新排队。

Runnable用于处理消息（使用joinPoint.procedure（）），然后将消息回复给Kafka。这是通过执行器在单独的线程中完成的。

我发现，如果在runnable中抛出异常，它会在catch语句中捕获，但当我们抛出新的PollableStreamListenerException异常时不会传播，这意味着失败不会在DLQ中结束。

我认为这是因为执行发生在另一个线程上，而主线程在执行过程中已经移动（不等待第二个线程处理消息），因此不再能够传播到DLQ。然而，我可能误解了这一点，因为我不太熟悉多线程的工作原理。

我已经尝试将executor.submit（runnable）修改为executor.submit（runnable）. get（），这解决了这个问题。但是它会导致阻塞主线程的执行，直到另一个线程完成执行，这意味着主线程不再可用于轮询新消息。实际上，这使我们的Kafka消费者不再是可污染的消费者，这违背了拥有注释的整个目的。

有人知道是否可以实现让主线程继续运行和轮询消息，同时在传播到DLQ的runnable中抛出异常吗？

提前感谢你的帮助。

为了提供更多的上下文，我们使用如下注释：

  @PollableStreamListener
  public void submitDeletion(Message<?> received) {
// Process message
}

每当在Pollable消息源上收到新消息时，都会调用submitDeletion方法。我们使用@附表检查新消息：

 @Scheduled(fixedDelayString = "${app.pollable-consumer.time-interval}")
  public void pollForDeletionRequest() {
    log.trace("Polling for new messages");
    cleanupInput.poll(cleanupSubmissionService::submitDeletion);
  }

更新：尝试使用CompletableFuture

根据@kriegaex的评论，我试图使用一个CompletableFuture。我简化了我的示例，使其更像一个POC。

@Aspect
@Component
@Slf4j
public class PollableStreamListenerAspect {

  private final ExecutorService executor = Executors.newFixedThreadPool(1);

  private volatile boolean paused = false;

  @Around(value = "@annotation(pollableStreamListener) && args(dataCapsule,..)")
  public void receiveMessage(ProceedingJoinPoint joinPoint,
      PollableStreamListener pollableStreamListener, Object dataCapsule) {
    if (dataCapsule instanceof Message) {
      Message<?> message = (Message<?>) dataCapsule;
      AcknowledgmentCallback callback = StaticMessageHeaderAccessor
          .getAcknowledgmentCallback(message);
      callback.noAutoAck();

      if (!paused) {
        CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {
          log.info("Start execution logging");
          try {
            Thread.sleep(10000);
          } catch (Exception e) {
            log.error("Error while sleeping", e);
          }
          log.info("End execution logging");
          throw new RuntimeException("Throwing exception to force handle statement");
        }, executor).handle((s, t) -> {
          log.info("Inside handle block:");

          if (t != null) {
            log.info(t.toString());
            throw new RuntimeException(t);
          }
          return null;
        });

        try {
          completableFuture.join();
        } catch (Exception e) {
          log.error("Error while doing join()", e);
        }

        callback.acknowledge(Status.ACCEPT);
      } else {
        // The separate thread is busy with a previous message, so re-queue this message for later:
        callback.acknowledge(Status.REQUEUE);
      }
    }
  }

}

我使用我的ExecutorService实例在CompletableFuture中运行函数。它抛出一个异常，该异常在中处理。handle（）block。

我的第一次测试没有以下代码行，我发现该函数在另一个线程上运行而没有阻止主线程进行轮询，但hand（）内部抛出的异常没有传播到DLQ。缺少代码行：

try {
          completableFuture.join();
        } catch (Exception e) {
          log.error("Error while doing join()", e);
        }

然后我在中添加了这些代码行，发现它在等待completableFuture完成运行时开始阻止主线程的执行。

简而言之，使用CompletableFuture的行为似乎与我在初始Runnable中发现的行为相同。




Answer 1

最后，我在Runnable中处理了这个错误，方法是手动将错误发布到DLQ，而不是依赖Spring Cloud Stream来为我处理这个问题。虽然这显然不是一个理想的解决方案，但最终的结果是有效的。