Kafka

Question

我无法批量阅读Kafka骆驼消费者，尽管遵循了这里发布的一个例子。我需要对我的生产者进行更改，还是我的消费者配置最有可能出现问题？

所讨论的应用程序利用kafka camel组件接收来自restendpoint的消息，验证它们，并将它们放在主题上。然后，我有一个单独的服务，从主题中使用它们，并将它们保存在时间序列数据库中。

消息是一次一个地产生和消费的，但是数据库希望消息是批量消费和提交的，以获得最佳性能。在不接触生产者的情况下，我尝试调整消费者以匹配这个问题答案中的示例：

如何从骆驼进行交易投票Kafka？

我不确定消息会如何显示，所以现在我只是记录它们：

    from(kafkaReadingConsumerEndpoint).routeId("rawReadingsConsumer").process(exchange -> {
        // simple approach to generating errors
        String body = exchange.getIn().getBody(String.class);
        if (body.startsWith("error")) {
            throw new RuntimeException("can't handle the message");
        }
        log.info("BODY:{}", body);
    }).process(kafkaOffsetManager);

但这些消息似乎仍然是一次只收到一条，而没有批量读取。

我的消费者配置如下：

  kafka:
    host: myhost
    port: myport
    consumer:
      seekTo: beginning
      maxPartitionFetchBytes: 55000
      maxPollRecords: 50
      consumerCount: 1
      autoOffsetReset: earliest
      autoCommitEnable: false
      allowManualCommit: true
      breakOnFirstError: true

我的配置是否需要工作，或者是否需要对制作人进行更改才能正确工作？

Answer 1

在最低层，Kafka消费者#投票方法将返回一个迭代器

我对Camel没有深入的经验，但为了获得“一批”记录，您需要一些中间收集来“排队”数据，最终将这些数据发送到下游的某个“收集消费者”流程。然后你需要一些“开关”处理器，上面写着“等待，处理这批”或“继续填充这批”。

就数据库而言，这个过程正是Kafka Connect JDBC Sink对batch.sizeconfig所做的。