使用 Apache Camel 的 Kafka 消费者

Question

我是 Apache 骆驼的新手。我们正在做 POC 以使用 Camel 开发 kafka 消费者。下面是示例代码。

    context.addRoutes(new RouteBuilder(){

      @Override
        public void configure() throws Exception {
            // TODO Auto-generated method stub

         from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}" 
                     + "&consumersCount={{consumer.consumersCount}}" 
                     + "&seekTo={{consumer.seekTo}}" 
                     + "&groupId={{consumer.group}}")
             .process(new Processor() {

                @Override
                public void process(Exchange exchange) throws Exception {

                    Message message = exchange.getIn();
                    Object data = message.getBody();

                    System.out.println(data);
                }
             })
             .to("seda:end");


  });

        context.start();

    ConsumerTemplate template=context.createConsumerTemplate();
    String info=template.receiveBody("seda:end",String.class);

    System.out.println(info);
}

我遇到以下问题：

1. 上下文在启动后立即停止。
2. 如果我使用消费者模板轮询到端点，它不会打印任何内容，而在 .process() 内部，当我在无限循环中启动上下文时，我能够打印 kafka 消息。为什么消费者模板无法打印。

Answer 1

1. 正如 Claus 已经评论的那样，您的 Camel 上下文将立即关闭，因为它没有阻塞。请参阅他评论中的链接。
2. 我认为您错过了 template.start(); 来启动您的消费者。有关示例，请参阅 this link。