如何实现Kafka消费者使用Spring-cloud-stream按需处理事件？

Question

我正在尝试实现一个PollableConsumer，当我在SpringBoot应用程序中遇到endpoint时，它会在特定条件下开始轮询来自Kafka的消息。

在特定条件下，我尝试了多种触发民意调查的方法，但显然，只有当它不断地从Kafka主题进行民意调查时，它才会起作用。（就像spring cloud stream文档中的所有示例一样）

我正在寻找这样的东西：

public interface CustomProcessor {
    @Input
    PollableMessageSource input();
}

 public void run() {
        boolean result = true;
        while (result) {
            result = input.poll(m -> {
                Event event = (Event) m.getPayload();
                GenericMessage<Event> genericMessage = new GenericMessage<>(event, m.getHeaders());
                eventMessageConsumer.consume(genericMessage);
            }, new ParameterizedTypeReference<Event>() {
            });

            try {
                Thread.sleep(1_000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
            if (result) {
                System.out.println("Success");
            }
        }
    }

当我碰到这样的endpoint时可能会触发：

@GetMapping("/process")
public void process() {
   SomeClass.run();
}

Answer 1

在您的代码示例中，我不确定输入了什么。poll返回，因为您将结果分配给布尔值，而这不是KafkaConsumer的结果。poll方法返回。

Kafka消费者可以根据需要暂停和重启，这不应该是问题。

来自最新的KafkaConsumer文档：

消耗流量控制

...

Kafka支持使用pause（Collection）和resume（Collection）对消费流进行动态控制，在将来的轮询（Duration）调用中分别暂停指定分配分区上的消费，并恢复指定暂停分区上的消费。

然后这来自暂停方法描述：

请注意，此方法不影响分区订阅。特别是，使用自动分配时，不会导致组重新平衡。

https://kafka.apache.org/22/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html

因此，在您的情况下，您可以简单地拥有一个易失性布尔值isPause标志，如果更改了该标志，您将根据需要发出暂停或恢复命令。Kafka消费者将完成剩下的工作。

另外，请注意，KafkaConsumer不是线程安全的，因此您应该在轮询数据的同一线程上发出这些命令。

Answer 2

显然，目前使用spring cloud stream不可能暂停PollableConsumer，所以我回到了基于事件的消息消费，并使用执行器来控制绑定的状态。遵循此线程和spring cloud stream文档，我注入了BindingsEndpoint并更改了绑定的状态，如下所示：

@RestController
public class EventController {
    @Autowired
    public EventController(BindingsEndpoint bindingsEndpoint) {
        this.bindingsEndpoint = bindingsEndpoint;
    }
    @GetMapping("/changeState")
    public void sendMessage(@RequestParam("state") String state) {
        if (state.equals("paused")) {
            bindingsEndpoint.changeState("MY_BINDING", 
                           BindingsEndpoint.State.PAUSED);
        }
        if (state.equals("resumed")) {
            bindingsEndpoint.changeState("MY_BINDING", 
                           BindingsEndpoint.State.RESUMED);
        }
    }

这并不是我想要实现的，但已经足够接近了。