如何用度量来计算IDEMTOTENT消费者的重复消息？

Question

我有一个Apache Camel应用程序，带有 。我需要一个具有重复消息总数的度量。我该如何实施这样的指标？

代码

@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/"));
      }
    };
  }
}

研究

我研究了 micrometerconstants ，但是我找不到重复的指标消息。

问题

如何用度量计算愿意消费者的重复消息数量？

Answer 1

我找到了解决方法，请参阅
幂等消费者：

如何处理路由中的重复消息
从 Camel 2.8 开始提供

您现在可以将 skipDuplicate 选项设置为 false，这会指示幂等使用者也路由重复消息。但是，通过将 Exchange 上的属性设置为 true，重复邮件已被标记为重复。我们可以通过使用基于内容的路由器或消息过滤器来检测这一点并处理重复的消息，从而利用这一事实。

例如，在下面的示例中，我们使用消息过滤器将消息发送到重复端点，然后停止继续路由该消息。

过滤重复消息

from("direct:start")
    // 指示幂等消费者不要跳过重复项，因为我们将过滤我们自己
    .idempotConsumer(header("messageId")).messageIdRepository(repo).skipDuplicate(false)
    .filter(属性(Exchange.DUPLICATE_MESSAGE).isEqualTo(true))
        // 通过将重复消息发送到其他地方来过滤掉它们，然后停止
        .to("模拟：重复")
        。停止（）
    。结尾（）
    // 这里我们只处理新消息（没有重复消息）
    .to("模拟：结果");

和MicrometerBuilders#micrometer：

默认 MicrometerEndpointBuilderFactory.MicrometerEndpointBuilder 千分尺（字符串路径）

Micrometer (camel-micrometer) 使用 Micrometer 库直接从 Camel 路线收集各种指标。类别：监控 自：2.22 Maven 坐标：org.apache.camel:camel-micrometer 语法：micrometer:metricsType:metricsName 路径参数：metricsType（必填） 指标类型 有 6 个枚举，取值可以可以是以下之一：COUNTER、GAUGE、LONG_TASK_TIMER、TIMER、DISTRIBUTION_SUMMARY、OTHER 路径参数：metricsName（必需） 的名称metrics 路径参数：tags 指标的标签

参数：
路径 - 指标类型：指标名称

退货：
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/"));
    }
  };
}