从多个用户线程访问 ProcessContext::forward

Question

给定：具有 KStream::transform 的 DSL 拓扑。作为 Transformer::transform 执行的一部分，从输入消息生成多条消息（可能是来自单个输入消息的数千条输出消息）。

根据从数据库中检索到的数据生成新消息。为了加快这个过程，我想创建多个用户线程来并行访问数据库中的数据。生成新消息后，线程将调用 ProcessContext::forward 向下游发送消息。

从不同线程调用 ProcessContext::forward 安全吗？

Answer 1

不安全，不允许从不同的线程调用 ProcessorContext#forward()。如果你尝试，将抛出异常。

作为一种解决方法，您可以让所有线程 "buffer" 它们的结果数据，并在下一次调用 process() 时收集所有数据。作为替代方案，您还可以安排一个标点符号来收集和转发来自不同线程的数据。