Http请求上的Spring Kafka监听器

Question

我正在使用 Spring 和 Kafka，我发出如下所示的 HTTP POST 请求，并通过 Kafka 主题将一些信息发送到另一个服务。

@RequestMapping(method = RequestMethod.POST, value = "/portfolio")
public void getPortfolio(
       Authentication auth,
    @RequestBody User user
) {
    //Data Transfer Object
    UserDTO dto = user.toDTO();
    dto.setId(((AuthenticatedUser) auth.getPrincipal()).getId());

    //Sending message to Kafka topic
    sender.sendPortfolioRequest(dto);
}

然后我想监听不同主题的响应并在 HTTP 响应中返回数据，但我被困在这里。我可以使用下面的监听器方法来监听响应，但不知道如何将两者放在一起。

@KafkaListener(
    topics = Topics.PORTFOLIO_RESULT,
    containerFactory = "portfolioKafkaListenerContainerFactory"
)
public void portfolioListener(UserPortfolioDTO portfolio) {
    System.out.println("Recieved Portfolio: " + portfolio.toString());
}

附：我是使用 HTTP 请求的新手，不知道这是否是实现我想要实现的目标的正确方法，或者我是否应该使用 POST 创建新资源并重定向到该资源或其他东西。

Answer 1

@KafkaListener 无法做到这一点，因为它是单独启动的，并且完全可以在自己的线程中运行。同时，您希望在 HTTP 请求线程中得到回复。

这里唯一可能的解决方案是 ConsumerFactory 和手动使用 Apache Kafka Consumer。所以，你发送，你从工厂获得一个Consumer实例，调用它的poll()被阻塞直到结果，构建一个HTTP响应并关闭Consumer。

Answer 2

据我了解，我建议您启用 asych http 请求以能够链接您的进程。

Creating Asynchronous Methods with springboot

此选项将允许您处理 sendPortfolioRequest 并释放 http 请求（否则您将收到来自客户端的 http 请求超时）。

您尝试做的事情看起来像是一种反模式：您想要链接一个 sychron http 请求（因为您的 http 客户端正在等待来自服务器的响应）和一个异步消息传递系统 (Kafka)。

为了能够做你想做的事，我建议你改变你的 http 端点并添加一个 websocket 以达到最佳实践。

请看Using WebSocket to build an interactive web application