如何在发布消息到broker的Java代码中添加Prometheus消息导出逻辑？

这个需求完全可以实现，Prometheus基于客户端埋点+服务端拉取的模式工作，你只需要在现有发布逻辑中加入指标埋点，再暴露指标端口给Prometheus采集即可，以下是Java生态下的完整实现步骤：

1. 引入相关依赖 #

如果是Maven项目，在pom.xml中加入Prometheus客户端依赖：

<!-- 原生Prometheus Java客户端依赖 -->
<dependency>
    <groupId>io.prometheus</groupId>
    <artifactId>simpleclient</artifactId>
    <version>0.16.0</version>
</dependency>
<dependency>
    <groupId>io.prometheus</groupId>
    <artifactId>simpleclient_common</artifactId>
    <version>0.16.0</version>
</dependency>

<!-- 如果是Spring Boot项目，推荐用Micrometer封装的Prometheus依赖，适配性更好 -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
    <version>1.10.0</version>
</dependency>

2. 确定需要统计的指标 #

消息发布场景下常见的可观测指标有这些，你可以按需选择：

• 消息发布总次数：带topic标签区分不同主题的发布量
• 消息发布结果统计：带status标签区分成功/失败次数
• 消息体大小分布：统计不同大小的消息占比
• 消息发布耗时：统计发布操作的延迟情况

3. 代码埋点实现 #

方案1：原生Prometheus Client埋点（适用所有Java项目） #

修改你的ModuleAMessagePublisher类，代码如下：

import io.prometheus.client.Counter;
import io.prometheus.client.Summary;

public class ModuleAMessagePublisher {

    @Inject
    @InternalBroker
    private MessagePublisher messagePublisher;

    // 定义消息发布计数器，带topic、status两个标签
    private static final Counter PUBLISH_COUNTER = Counter.build()
            .name("modulea_mq_publish_total")
            .help("ModuleA发布到MQ的消息总次数")
            .labelNames("topic", "status")
            .register();

    // 定义消息大小统计指标
    private static final Summary MESSAGE_SIZE_SUMMARY = Summary.build()
            .name("modulea_mq_message_size_bytes")
            .help("ModuleA发布的消息体大小")
            .labelNames("topic")
            .register();

    public void publish(String topic, final String message) {
        log.info("<><><><><><><><><> (MQ) PUBLISH MODULEA MESSAGE: <><><><><><><>\n", message);
        // 埋点：统计消息大小
        MESSAGE_SIZE_SUMMARY.labels(topic).observe(message.getBytes().length);
        try {
            messagePublisher.publish(topic, message);
            // 埋点：发布成功计数+1
            PUBLISH_COUNTER.labels(topic, "success").inc();
        } catch (Exception e) {
            // 埋点：发布失败计数+1
            PUBLISH_COUNTER.labels(topic, "fail").inc();
            throw e;
        }
    }
}

方案2：Spring Boot项目用Micrometer埋点（更推荐） #

如果是Spring Boot项目，直接用框架自动注入的MeterRegistry做埋点，无需手动管理指标注册：

import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class ModuleAMessagePublisher {

    @Autowired
    @InternalBroker
    private MessagePublisher messagePublisher;

    @Autowired
    private MeterRegistry meterRegistry;

    public void publish(String topic, final String message) {
        log.info("<><><><><><><><><> (MQ) PUBLISH MODULEA MESSAGE: <><><><><><><>\n", message);
        // 埋点：统计消息大小
        meterRegistry.summary("modulea.mq.message.size.bytes", "topic", topic)
                .record(message.getBytes().length);
        try {
            messagePublisher.publish(topic, message);
            // 埋点：发布成功计数+1
            meterRegistry.counter("modulea.mq.publish.total", "topic", topic, "status", "success")
                    .increment();
        } catch (Exception e) {
            // 埋点：发布失败计数+1
            meterRegistry.counter("modulea.mq.publish.total", "topic", topic, "status", "fail")
                    .increment();
            throw e;
        }
    }
}

4. 暴露指标给Prometheus采集 #

• 非Spring Boot项目：引入simpleclient_httpserver依赖，在项目启动时启动一个内置HTTP服务，暴露/metrics端点即可
• Spring Boot项目：在application.properties中打开actuator的Prometheus端点：

management.endpoints.web.exposure.include=prometheus
management.endpoint.prometheus.enabled=true

配置完成后访问http://你的服务IP:端口/actuator/prometheus就可以看到刚才埋点的指标数据。

5. 配置Prometheus采集规则 #

在Prometheus的配置文件prometheus.yml中新增采集任务，指向你的服务端点：

scrape_configs:
  - job_name: "modulea-mq-publisher"
    scrape_interval: 15s
    static_configs:
      - targets: ["你的服务IP:端口"]

重启Prometheus后，就可以在Prometheus UI中查询、可视化你埋点的指标，也可以基于指标配置告警规则。

注意：埋点的标签基数不要过大，比如不要把消息内容、动态ID作为标签值，避免造成Prometheus存储压力过高。

内容的提问来源于stack exchange，提问作者R0b0t0