如何自动调用 kafka-stream-processor 中的 process() 方法

Question

我正在学习 kafka 流，并编写了一个简单的应用程序，片段如下：

主应用：

        Topology topology = new Topology();
        topology.addSource("SOURCE", "source-topic");
        topology.addProcessor("Processor1", () -> new Processor1(), "SOURCE");
        topology.addProcessor("Processor2", () -> new Processor2(), "Processor1");
        topology.addProcessor("Processor3", () -> new Processor3(), "Processor2");
        topology.addSink("SINK", "sink-topic", "Processor3");
        KafkaStreams streams = new KafkaStreams(topology, config);
        streams.start();

单个流处理程序的代码段：

public class Processor1 implements Processor<String, String> {
   // Rest of code
    @Override
    public void process(String key, String value) {
        System.out.println("Inside Processor1#process() method");
        context.forward(key, value);
    }

我知道我们需要创建Topology然后启动它，我们调用streams.start();

我无法理解该方法是如何自动调用process()以及谁调用它？

Answer 1

Processor process()由

ProcessorContextImpl类在特定拓扑节点的每个传入消息上自动调用的方法。对于构建的拓扑，当消息到达传入主题时，SOURCE节点会使用它，并通过内部调用forward方法将消息转发(传播(到子节点(您可以调试/查看类 ProcessorContextImpl 中的代码(。在您的情况下，SOURCE节点将键和值转发到子节点Processor1。之后，触发类Processor1 process()方法。当代码到达 context.forward() 时，消息转发到下一个子节点，Processor2 。之后，该消息以类似的方式传播到Processor3和SINK节点，最后，将消息生成到出站主题。特定消息的此类管道在单个线程上执行(如果您有 config num.stream.threads = 1 的默认值，则所有消息都将在每个应用实例的单个线程上处理(。