如何用度量来计算Idempotent使用者的重复消息？

Question

我有一个带有幂等消费者的Apache应用程序。我需要一个包含重复消息总数的度量。我怎样才能实行这样一个指标呢？

码

@SpringBootApplication
public class TestApplication {

  public static void main(String[] args) {
    SpringApplication.run(TestApplication.class, args);
  }

  @Bean
  public MicrometerRoutePolicyFactory micrometerRoutePolicyFactory() {
    return new MicrometerRoutePolicyFactory();
  }

  @Bean
  public EndpointRouteBuilder route() {
    return new EndpointRouteBuilder() {
      @Override
      public void configure() throws Exception {
          from(file("d:/tmp/camel/"))
             .idempotentConsumer(jsonpath("$.id"), MemoryIdempotentRepository.memoryIdempotentRepository())
             .to(file("d:/tmp/copy/"));
      }
    };
  }
}

Research

我查看了MicrometerConstants，但找不到重复消息的度量标准。

问题

如何用度量来计算的重复消息数量？

Answer 1

我找到了一个解决办法，见幂等消费者

如何在Camel 2.8可用的路由中处理重复消息 现在可以将skipDuplicate选项设置为false，这指示幂等用户也路由重复消息。但是，通过将Exchange上的属性设置为true，重复消息已被标记为重复。我们可以利用这个事实，使用基于内容的路由器或消息筛选器来检测这一点并处理重复的消息。 例如，在下面的示例中，我们使用消息筛选器将消息发送到重复的端点，然后停止继续路由该消息。 过滤器重复消息 从(“直接:开始”) //指示幂等消费者不要跳过重复信息，因为我们将过滤我们自己的.idempotentConsumer(header("messageId")).messageIdRepository(repo).skipDuplicate(false) .filter(property(Exchange.DUPLICATE_MESSAGE).isEqualTo(true)) //过滤出重复消息，将它们发送到其他地方，然后停止.to(“模拟:重复”) .stop() .end()//这里我们只处理新消息(没有重复消息).to(“模拟:结果”)；

和MicrometerBuilders#micrometer

default MicrometerEndpointBuilderFactory.MicrometerEndpointBuilder micrometer(String path) 千分尺(骆驼-千分尺)使用千分尺库直接从骆驼路线收集各种指标。类别:监视自: 2.22 Maven坐标:org.apache.cammel:骆驼-微米语法：micrometer:metricsType:metricsName路径参数: metricsType (必需)度量类型有6个枚举，其值可以是:计数器、量规、LONG_TASK_TIMER、计时器、DISTRIBUTION_SUMMARY、其他路径参数:度量路径参数的metricsName (必需)名称:度量标记 参数： path - metricsType:metricsName 返回： dsl构建器

我修改的代码：

@Bean
public EndpointRouteBuilder route() {
  return new EndpointRouteBuilder() {
    @Override
    public void configure() throws Exception {
       from(file("d:/tmp/camel/"))
         .idempotentConsumer(jsonpath("$.id"), MemoryIdempotentRepository.memoryIdempotentRepository())
           .skipDuplicate(false)
         .filter(header(Exchange.DUPLICATE_MESSAGE).isEqualTo(true))
           .to(micrometer("duplicate_messages").increment("1"))
           .stop()
         .end()
         .to(file("d:/tmp/copy/"));
    }
  };
}