Spring Kafka Listener по запросу Http

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());
}

P.S. Я новичок в использовании HTTP-запросов и не знаю, является ли это правильным способом сделать то, что я пытаюсь достичь, или мне следует создавать новый ресурс с POST и перенаправлять на это или что-то еще.

Answer 1

Согласно тому, что я понимаю, я предлагаю вам включить asych http-запрос, чтобы иметь возможность связать ваш процесс.

Создание асинхронных методов с помощью Springboot

Эта опция позволит вам обработать sendPortfolioRequest и освободить http-запрос (в противном случае у вас будет таймаут http-запроса со стороны клиента).

И то, что вы пытаетесь сделать, выглядит как анти-шаблон: вы хотите связать http-запрос sychron (потому что ваш http-клиент ожидает ответа от сервера) и асинхронную систему обмена сообщениями (Kafka).

Чтобы иметь возможность делать то, что вы хотите, я предлагаю вам изменить конечную точку http и добавить веб-сокет, чтобы это было наилучшим способом.

См. Использование WebSocket для создания интерактивного веб-приложения

Answer 2

Этого нельзя сделать с @KafkaListener, так как он запускается отдельно и полностью работает в своих собственных потоках. Между тем вы ожидаете ответа в потоке HTTP-запроса.

Единственное возможное для вас решение - это ConsumerFactory и ручное использование Apache Kafka Consumer. Итак, вы отправляете, вы получаете Consumer экземпляр с фабрики, вызываете его poll() заблокированным до результата, строите ответ для HTTP и закрываете Consumer.