如何使用 Micrometer 获取 Spring Integration 队列中的消息数？

Posted 2023-02-15

【中文标题】如何使用 Micrometer 获取 Spring Integration 队列中的消息数？

【英文标题】：How do I get the number of messages in a Spring Integration queue with Micrometer?

【发布时间】：2020-11-17 14:14:11

【问题描述】：

Spring 的Metrics and Management: MessageChannel Metric Features 描述：

如果是QueueChannel，您还会看到接收操作的统计信息以及当前由该QueueChannel 缓冲的消息计数

但是：

这些旧指标将在未来的版本中移除。见Micrometer Integration。

描述的地方：

可轮询消息通道上用于接收操作的计数器计量器具有以下名称或标签： name: spring.integration.receive [...]

这听起来像是只计算已收到多少条消息。队列中的消息数量似乎不可用，即使通过计算receive - send（因为没有send）。

那么，使用 Spring Integration 和 Micrometer，甚至可以读取队列大小吗？怎么样？

【参考方案1】：

我认为我们需要为这个队列大小值注册一个Gauge：

/**
 * A gauge tracks a value that may go up or down. The value that is published for gauges is
 * an instantaneous sample of the gauge at publishing time.
 *
 * @author Jon Schneider
 */
public interface Gauge extends Meter 

这是一个相应的构建器：

/**
 * A convenience method for building a gauge from a supplying function, holding a strong
 * reference to this function.
 *
 * @param name The gauge's name.
 * @param f    A function that yields a double value for the gauge.
 * @return A new gauge builder.
 * @since 1.1.0
 */
@Incubating(since = "1.1.0")
static Builder<Supplier<Number>> builder(String name, Supplier<Number> f) 

随时提出 GH 问题来改进！

目前可能通过带有QueueChannel.getQueueSize() 委托的外部Gauge 实例。

【讨论】：

这里是修复：github.com/spring-projects/spring-integration/pull/3349

你们超级快速且乐于助人，非常感谢！ :-) 非常感谢。

感谢您确认修复是正确的！感谢您的评论！

【参考方案2】：

在 Spring Integration 5.4 发布之前，可以使用这个：

import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.management.metrics.MetricsCaptor;
import org.springframework.integration.support.management.micrometer.MicrometerMetricsCaptor;

import java.util.Collection;

@Configuration
@RequiredArgsConstructor
public class IntegrationMetricConfig 

  private final ApplicationContext applicationContext;
  private final Collection<QueueChannel> queues;

  /**
   * Temporary solution until <a href="https://github.com/spring-projects/spring-integration/pull/3349/files">Add
   * gauges for queue channel size</a> is released. Remove this when updating to Spring Integration
   * 5.4.
   */
  @Autowired
  public void queueSizeGauges() 
    MetricsCaptor metricsCaptor = MicrometerMetricsCaptor.loadCaptor(this.applicationContext);
    queues.forEach(queue -> 
      metricsCaptor.gaugeBuilder("spring.integration.channel.queue.size", queue,
        obj -> queue.getQueueSize())
        .tag("name", queue.getComponentName() == null ? "unknown" : queue.getComponentName())
        .tag("type", "channel")
        .description("The size of the queue channel")
        .build();

      metricsCaptor.gaugeBuilder("spring.integration.channel.queue.remaining.capacity", this,
        obj -> queue.getQueueSize())
        .tag("name", queue.getComponentName() == null ? "unknown" : queue.getComponentName())
        .tag("type", "channel")
        .description("The remaining capacity of the queue channel")
        .build();
    );