Apache Camel Kafka-聚合Kafka消息并定期发布到不同的主题

Question

我有一个用例：

我需要定期阅读和汇总Kafka主题的信息，并发布到不同的主题。本地存储不是一个选项。这就是我计划解决这个问题的方式，欢迎提出任何改进建议

为了安排Kafka消息的聚合和发布，计划使用聚合器EIP的completionInterval选项。这是代码。

  @Autowired ObjectMapper objectMapper;
  JacksonDataFormat jacksonDataFormat;

  @PostConstruct
  public void initialize(){
    //objectMapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
    jacksonDataFormat = new JacksonDataFormat(objectMapper,EventMessage.class);
  }

路线是：

public void configure() throws Exception {
    from("kafka:localhost:9092?topic=item-events" +
            "&groupId=aggregator-group-id&autoCommitIntervalMs=25000&autoOffsetReset=earliest&consumersCount=1")
            .routeId("kafkapoller")
            .unmarshal(jacksonDataFormat)
            .aggregate(body().method("getItemId"), new EventAggregationStrategy()).completionInterval(20000)
            .marshal().json(JsonLibrary.Jackson)
            .to("kafka:localhost:9092?topic=item-events-aggregated&serializerClass=org.apache.kafka.common.serialization.ByteArraySerializer");
  }

Answer 1

这看起来不错。要记住的事情：

• 如果/当JVM在聚合周期中途死亡时会发生什么？不要在意，然后冷静，否则您可能需要研究PeristentAggregationRepository来存储/重放消息，尽管您可以重放从kafka丢失的消息（这将是我最大的操作问题）
• 接下来，考虑运行时控制。骆驼是一个令人震惊的人，因为它没有非常清楚地告诉你运行时发生了什么。聚合器（即非常贪婪的正则表达式）中的失控方法之类的东西会让你对聚合交换的当前状态知之甚少，JMX可能不会告诉你太多正在发生的事情。
• 我将使用Aggregate Controller使您能够外部强制完成交换，因此您可以执行诸如向骆驼发出关闭命令，然后调用此命令来完成机上交换