一、 背景与目标
监控最主要还是上报到Prometheus，可惜成本实在是高昂，特别是存储时间要求得越久，我们这里探索了下micrometer转换成json，然后存储到starrocks这类mpp olap引擎，通过starrocks的存算分离架构，降低成本。
二、 总体架构
Spring Boot + Micrometer --指标采集--> Kafka --消息流--> Flink --清洗/聚合/分流--> StarRocks (metric_data / metric_series) 三、Spring Boot MeterRegister 发送至kafka
在此之前，我们上报Prometheus的时候，一般都是再Spring Boot中配置micrometer-registry-prometheus即可，这种方式主要是Pull模式，暴露/actuator/prometheus的指标，然后由Prometheus抓取。同时还有一种Step模式，即：定义一个周期T，周期结束时相关指标清零。
| 特性 | Step（Delta）模式 | Cumulative（累计）模式 |
|---|---|---|
| 值含义 | 当前时间窗的增量 | 从启动至今的总和 |
| 是否清零 | 是（周期切换后清零） | 否 |
| 场景 | Push 型采集系统 | Pull 型采集系统 |
| 数据延迟影响 | 延迟会导致丢失当前周期数据 | 延迟只影响分辨率，不丢数据 |
| 计算速率方式 | 注册表已按步长计算 | 采集端函数计算 |
| 如果采用累计模式，存储到数据库里则是不断增加的值，如果要计算某一段分钟或者小时的值，则需要使用窗口函数，对数据库压力巨大。相比之下，直接在采集的时候就做好差分，对数据库压力很小，故Step步长模式更适合我们。 |
3.1 代码样例
先是定义export的配置，也可以做成通用的spring-boot-starter给别的服务用。
@Data @Component @ConfigurationProperties(prefix = "management.metrics.export.kafka") public class KafkaMeterRegistryConfig implements StepRegistryConfig { private String topic = "default_topic"; private Duration step = Duration.ofSeconds(30); @Override @NonNull public String prefix() { return "management.metrics.export.kafka"; } @Override public String get(@NotNull String key) { return null; } @Override @NonNull public Duration step() { return step; } } 其次，通过扩展定义您自己的计量表注册表StepMeterRegistry，如下所示：
public class KafkaMeterRegistry extends StepMeterRegistry { private final KafkaTemplate<String, String> kafkaTemplate; private final String topic; public KafkaMeterRegistry(KafkaMeterRegistryConfig config, Clock clock, KafkaTemplate<String, String> kafkaTemplate) { super(config, clock); this.kafkaTemplate = kafkaTemplate; this.topic = config.getTopic(); start(new NamedThreadFactory("kafka-metrics-publisher")); log.info("KafkaMeterRegistry created"); } @Override protected void publish() { // 每个step结束后该方法被调用 long ts = System.currentTimeMillis(); for (Meter meter : this.getMeters()) { //关键字：窗口切换 //不要在 publish() 里重复调用同一个 meter.measure() 多次，否则第一次调用就清零了，后面会拿到零值。 //不要依赖 publish() 之后还去别处用同一个 StepMeterRegistry 的 Meter 做跨周期累计，因为 step 运行就是为时间窗统计设计的，跨周期值会被清零。 for (Measurement ms : meter.measure()) { ...准备数据 kafkaTemplate.send(topic, jsonString); } } } @NotNull @Override protected TimeUnit getBaseTimeUnit() { return TimeUnit.MILLISECONDS; // 发送数据时间单位 } } 最后，创建注册表配置和计量表注册表相关的BEAN。如果您使用 Spring Boot，可以按如下方式操作：
@Configuration public class MetricsConfig { @Bean public KafkaMeterRegistry customMeterRegistry(KafkaMeterRegistryConfig kafkaMeterRegistryConfig, Clock clock, KafkaTemplate<String, String> kafkaTemplate) { return new KafkaMeterRegistry(kafkaMeterRegistryConfig, clock, kafkaTemplate); } } 3.2 底层如何实现Step模式
目前，核心疑问点在于是如何清零的？是否有丢数据的可能？
在 Micrometer 中，步长周期（Step Interval）的处理逻辑主要由 StepMeterRegistry 完成，这类 Registry（例如 DatadogMeterRegistry、AtlasMeterRegistry）会确保每个时间窗口的指标值是该窗口的增量数据，而不是自应用启动以来的累计值。
┌─ StepMeterRegistry.publish() ← 定时任务 │ ├─ meter.measure() ← 对 StepCounter 调用 measure │ ├─ StepCounter.poll() │ ├─ double v = count.get() │ ├─ count.set(0.0) ← 清零 │ ├─ lastStepTime.set(now) ← 标记窗口开始时间 │ └─ 返回当前窗口的值给 publish 简单的说，就是：
StepMeterRegistry定时（按 step 配置）调用publish()。publish()会遍历所有的 meter，包括 Counter。- 对 StepCounter 而言，
poll()会返回当前窗口增量并清零。 - 下一个步骤周期继续从 0 开始累加。
因此，清零是在采集周期触发，其条件是：
now - lastStepTime >= stepMillis
处理完后发送到kafka就是类似：
{ "ts": "2024-06-01 12:00:00.000", "name": "http_server_requests_total_time", "env": "prod", "labels": { "uri": "/api/orders", "method": "POST" }, "value": 3.14159 } 四、 存储方案对比
我们以一串json为例
{ "metricName": "cpu_usage", "labels": { "host": "server-01", "region": "us-east" }, "timestamp": "2025-10-15T08:01:12.123Z", "value": 0.73 } 直接可以对第一层key建表，metricName、timestamp、value都可以直接约定为列，目前问题的核心主要有两个：
（1）labels如何存储
（2）针对Histogram、Summary、Gauge的上报如何处理。
我们先来解决第一个问题，labels如何存储。
4.1 方案一：字符串拼接
如果维度比较少的情况下，确实可以考虑通过字符串截取来实现，比如上面的存储为server-01|us-east，通过split竖线来获取响应的字段，例如：
SPLIT_PART('server-01|us-east', '|', 1) AS host_value, SPLIT_PART('server-01|us-east', '|', 2) AS region_value 这种如果labels多的情况下，需要自行约定字符串的顺序，而且数据量的情况下，SPLIT_PART性能消耗也比较大。
4.2 方案二：指标表 + 标签维表 + 倒排索引
这种主要是想把指标和labels拆出来放到不同的两个表里，同时使用一个唯一id关联为一批数据，一开始有点豁然开朗，后来一想，（1）维度如果很多，但是只查一个维度的数据，这种关联不准确；（2）如果要关联查询，特别是维度多的情况下，sql复杂度翻好几倍，维护性大大降低。
4.3 方案三：Flat JSON 存储
考虑了一下简单粗暴的方法，labels直接存储为JSON列，刚好看到starrocks的新特性Flat JSON，Flat JSON的核心原理是在导入时检测JSON数据，并从JSON数据中提取常用字段，作为标准类型数据存储。在查询JSON时，这些常用字段优化了JSON的查询速度，刚好符合我们的条件。
CREATE TABLE monitor_data ( ts DATETIME COMMENT '指标时间', name STRING COMMENT '指标名称', env STRING COMMENT '环境', dc STRING COMMENT '数据中心', biz_id BIGINT COMMENT '业务 ID', labels JSON COMMENT '指标标签（JSON 格式）', value DOUBLE COMMENT '指标值' ) ENGINE = OLAP DUPLICATE KEY(`ts`,`name`,`env`,`dc`,`biz_id`) COMMENT "监控" PARTITION BY date_trunc('day', ts) DISTRIBUTED BY HASH(`name`,`env`,`dc`,`biz_id`) ORDER BY (`name`,`ts`) PROPERTIES ( "compression" = "LZ4", "flat_json.enable" = "true", "fast_schema_evolution" = "false", "partition_live_number" = "366", "replicated_storage" = "true", "replication_num" = "3" ); 五、Flink入库
starrocks是可以直接通过jdbc方式入库的，但是JDBC 适合低并发、小批量，容易出现性能问题，对于复杂多表、多源、字段更新场景，Kafka + Flink + StarRocks 是最安全高效的方案。具体也不展开讲解，这个配合下公司基建来即可。
六、入库后样例
- 写入数据后展示：
| ts | name | env | dc | biz_id | labels | value |
|---|---|---|---|---|---|---|
| 2024-06-01 12:00:00 | http_requests_count | prod | eu | 0 | 3.14159 |
- 最近 5 分钟各接口请求数（以 http_requests_count 为例，按 method/status 维度聚合）：
SELECT JSON_VALUE(labels, '$.method') AS method, JSON_VALUE(labels, '$.status') AS status, SUM(value) AS total_requests FROM monitor_data WHERE name = 'http_requests_count' AND ts >= NOW() - INTERVAL 5 MINUTE AND env = 'prod' GROUP BY method, status ORDER BY total_requests DESC; 
