本文介绍实时数据订阅功能的数据消费格式定义说明和示例。

前提条件

您已经完成客户端配置（kafka/Prometheus监控系统/api地址），且公网网络联通正常。同时客户端服务器入方向端口允许访问。

说明
本文以kafka消费数据为例，基本操作步骤如下：
安装kafka客户端，并修改相关配置信息。
配置客户端服务器网络入方向允许访问。
启动客户端kafka程序。
在云监控控制台，创建数据订阅任务。（填写客户端kafka相关配置信息）

数据消费定义说明

告警数据不做压缩，直接消费即可
监控数据vminsert写入kafka的数据采用zstd压缩，单条message可能包含多个metrics指标，消费到数据后先解压缩，解压后生成数据如下：

{

	"metrics":[{

		"tags":{

			"__report_by__":"harvest",

			"cpu":"cpu-total",

			"host":"test-2dd4bfrv",

			"idc":"neimengaz03",

			"job":"virtual_machine",

			"__name__":"cpu_util",

			"uuid":"test-8194e630-fec1"

		},

		"fields":{

			"cpu_util":0.26697814117055
		},

		"name":"prometheus_remote_write",

		"timestamp":1713482161

	}]

}

告警数据

type AlarmSubscription struct{

	Service   string `json:"service"`   //服务

	Dimension string `json:"dimension"` //维度

	RegionId  string `json:"region_id"` //Region Id

	Idc       string `json:"idc"`       //资源池标识符

	Key       string `json:"key"`       //账号ID


	ModelId string `json:"model_id"` //控制台告警规则Id

	IssueId string `json:"issue_id"` //IssueID， 这个字段方便问题回溯

	InfoId  string `json:"info_id"`  //InfoID，  这个字段方便问题回溯


	Name      string `json:"name"`       //规则名称

	AlarmType string `json:"alarm_type"` //告警类型：series: 时序类监控、event: 事件类

	Status    int    `json:"status"`     //告警状态：0：告警 1: 恢复


	Ctime    int64       `json:"ctime"`    //告警首次发生时间

	Value    interface{} `json:"value"`    //告警当前值，  告警恢复时，不存在该字段

	Resource []Object    `json:"resource"` //告警实例信息


	Metric    string `json:"metric"`     //数据指标， 告警类型为event时，该字段为空

	AlarmName string `json:"alarm_name"` //告警规则名称

	Threshold string `json:"threshold"`  //告警阈值

	Operator  string `json:"operator"`   //告警阈值比较符

	Unit      string `json:"unit"`       //告警数据单位

}

消费数据乱码问题

对于Java生产者和消费者，可以在配置中设置字符编码：

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("serializer.encoding", "UTF-8"); // 设置编码格式
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("deserializer.encoding", "UTF-8"); // 设置编码格式
如果使用的是命令行工具，确保控制台的字符编码与Kafka消息编码一致。

在Windows系统中，可以通过修改控制台默认编码为UTF-8来解决：

chcp 65001
在Linux系统中，可以通过设置环境变量来调整字符编码：
export LANG=en_US.UTF-8