Kafka 反应器响应

Question

我的团队正在开发一个聊天应用程序，该应用程序使用 Kafka Reactor 在客户端之间传输数据。

我们的问题之一是向用户返回反馈。通过调用像 ex 这样的 API。 “/registerUser” 人们希望从同一端点收到注册过程是否成功的反馈，而不必调用另一个端点来找出结果，这就是 kafka 处理事情的方式。

我是使用 kafka 的新手，刚刚开始对它进行猛烈抨击。我是在尝试做一些不可能的事情还是我看错了。

如有任何帮助，我们将不胜感激。预先感谢。

Answer 1

让我帮助解释如何在聊天应用程序中使用 Kafka Reactor 有效处理这种情况。

您想要实现的目标绝对是可能的。这是使用 Spring WebFlux 和 Kafka Reactor 的解决方案，可提供即时反馈：

@RestController

@RequestMapping(“/api”) 公共类 UserController {

@Autowired
private KafkaTemplate<String, UserRegistrationEvent> kafkaTemplate;

@PostMapping("/registerUser")
public Mono<ResponseEntity<RegistrationResponse>> registerUser(@RequestBody UserRegistration request) {
    return Mono.create(emitter -> {
        // Create a correlation ID for this registration
        String correlationId = UUID.randomUUID().toString();
        
        // Create registration event
        UserRegistrationEvent event = new UserRegistrationEvent(
            correlationId,
            request.getUsername(),
            request.getEmail()
        );
        
        // Send to Kafka and listen for response
        kafkaTemplate.send("user-registration", correlationId, event)
            .addCallback(
                success -> {
                    // Registration sent successfully
                    emitter.success(ResponseEntity.ok(
                        new RegistrationResponse(correlationId, "Registration in progress")
                    ));
                },
                failure -> {
                    // Failed to send registration
                    emitter.error(new RuntimeException("Failed to process registration"));
                }
            );
    });
}

}