数据目的地
时序数据库(TSDB)
如需视频教程,请点击这里。
写时序数据可以划分成两个步骤:
- 对消息依据SQL进行过滤和变换
- 根据结果消息以指定的方式写TSDB
一个输入MQTT消息经过第一步的过滤和变换后的结果消息,如果符合下述几种格式中的任何一种,或者多种,则消息会被写进TSDB。如果一个结果消息符合多种格式,则会以多种方式并行地写TSDB。
每种格式,都需要在结果消息中明确的指定如下几个参数:
- 度量(metric),一个TSDB下面可以有多个度量,一个度量相当于MySql数据库中的表
- 时间戳,每个数据点写入TSDB均需要明确地指定时间戳。部分格式可以指定输入消息时间戳的格式,其他情况下,系统会自动识别下述几种常见的时间戳格式
- 域,如果不是默认域(value)的话,需要指定将数据存入哪个域。一个度量下面可以有多个域,一个域相当于MySql表中的字段
- 值,就是写进TSDB的真正的数据。值分数值型、字符串型、以及byte型。特别说明,如果结果消息本身是字符串类型,但是可以理解成数值类型,那么可能优先以数值类型写。同一个度量同一个域,写的数据类型不能变
- 一个或者多个tag, tag用来区分同一个metric下的数据,方便从TSDB中检索数据。例如,为了区分device_temperate度量下,温度是属于那个设备的,可以增加一个名为device的tag,tag的值为每个设备的编号
时序数据库的metric, tag和域均有相应的命名规范。
为了使数据符合TSDB的命名要求,规则引擎在如下几种情况,对metric, field name, tag name, tag value做如下默认的纠正:
- metric:长度不够时,补"_",超长部分,截断;
- field name:长度不够时,补"_",超长部分,截断;
- tag key:非法字符替换成"_";首字符不为字母,补"L";超长部分,截断;
- tag value:长度不够时,补"_",超长部分,截断;
系统能识别的常用时间戳格式如下所示:
{"ts":1523427241000}
{"ts":1523427241}
{"ts":"1523427241000"}
{"ts":"1523427241"}
{"ts":"2018-03-25T07:05:26.847Z"}
{"ts":"2018-03-25 07:05:26.847Z"}
{"ts":"2018-03-25T07:05:26Z"}
{"ts":"2018-03-25 07:05:26Z"}
{"ts":"2018-03-25T07:05:26.847"}
{"ts":"2018-03-25 07:05:26.847"}
{"ts":"2018-03-25T07:05:26"}
{"ts":"2018-03-25 07:05:26"}
根据消息格式的不同,以及是否写多域,写时序数据库分如下几种模式。
单点(默认域)
该方式将您的一个MQTT消息转化成一个点,写到某个度量(metric)中的默认域(value)中去。经过SQL变换后的消息,应该包含如下几个字段
- metric: 指明数据写到TSDB的度量名
- value: 数据部分,该字段也可以用'_value'代替
- timestamp: 写入时序数据库的时间戳,该字段也可以用'_timestamp'代替
- 一个或者多个tag: 任意名字,key作为tag的名字,value作为tag的值。
举例
结果消息的格式应该类似如下:
{
"metric": "device_temperature",
"value": 35.6,
"timestamp": 1534975332,
"deviceid": "dev001",
"factory": "shanghai"
}
上述消息中,deviceid, factory均作为tag。
在实际中,设备上传的消息很可能不符合上述格式,此时可以通过查询字段的SQL语句来将消息变换成合法的格式。假如设备上传消息如下:
{
"temperature": 35.6,
"ts": 1534975332,
"deviceid": "dev001",
"factory": "shanghai"
}
则需要类似如下的SQL来转换:
'device_temperature' AS metric, temperature AS `value`, ts AS `timestamp`, deviceid, factory
'device_temperature' AS metric: 在结果消息中新增一个名为metric的字段,值为字符串'device_temperature';
temperature AS `value`: 通过AS语法,将temperature字段重命名成value;注意,value是SQL中的保留字,因此需要用重音符(back tick)引起来;这一句也可以写成: temperature AS _value, _value不是保留字,因此不需要重音符;
ts AS `timestamp`: 将ts重命名成timestamp。注意,timestamp也是保留字,因此也需要重音符引起了。这句也可以写成:ts AS _timestamp;
deviceid: 将输入消息中的deviceid SELECT过来,名字不变。
factory: 将输入消息中的factory SELECT过来,名字不变。
数组(默认域)
该方式适合处理待写入数据位于JSON数组的情况,其他字段如metric, timestamp, tag可以在数组内,也可以在数组外。数据写入默认域(value)。
该方式需要在结果消息中包含一个名为"_TSDB_META"的字段,描述以何种方式写TSDB。该字段描述了:
- data_array: 待处理数组的JSON路径
- value_field: 数据在数组中的相对路径
- global_metric或者point_metric: metric的路径,如果metric在数组中,则用point_metric表示,每个点使用不同的metric;如果在数组外,则用global_metric表示,所有点使用同一个metric
- global_time或者point_time: 时间戳的路径,如果时间戳在数组中,则用point_time表示,每个点使用不同的时间戳;如果在数组外,则用global_time表示,所有点使用相同的时间戳
- global_tags和point_tags: tag的路径,如果tag在数组中,则用point_tags.tagX表示,每个点使用不同的tag;如果在数组外,则用global_tags.tagX表示,所有点使用相同的tag。其中tagX中的X为一个数字下标,如tag1, tag2, tag3等,不同的tag用不同的下标。当然,如果数据点既有独立的tag也有共同的tag,那么也可以point_tags和global_tags共存
- 时间戳格式(可选),如果时间戳为非标准格式,可以指定时间戳的格式,方便系统识别时间戳。在_TSDB_META.time_format字段中指定时间戳格式,如:"yy-MM-dd HH:mm:ss"。
举例
结果消息的格式应该类似如下:
{
"data": {
"arr":[{
"val": 23.4,
"ts": 1534975332
},{
"val": 34.1,
"ts": 1534975337
}]
},
"deviceid": "device002",
"factory": "shanghai",
"name": "mymetric",
"_TSDB_META": {
"data_array":"data.arr",
"value_field":"val",
"point_time":"ts",
"global_metric":"name",
"global_tags":{
"tag1": "deviceid",
"tag2": "factory"
}
}
}
假如设备上传的消息格式如下:
{
"data": {
"arr": [{
"val": 23.4,
"ts": 1534975332
}, {
"val": 34.1,
"ts": 1534975337
}]
},
"deviceid": "device002",
"factory": "shanghai"
}
希望将data.arr这个数组的每个元素,val作为值,ts作为时间戳写进TSDB,则可以通过如下的SQL语句,使之符合数组的写入格式:
*, 'mymetric' AS name, 'data.arr' AS _TSDB_META.data_array, 'val' AS _TSDB_META.value_field, 'ts' AS _TSDB_META.point_time, 'name' AS _TSDB_META.global_metric, 'deviceid' AS _TSDB_META.global_tags.tag1, 'factory' AS _TSDB_META.global_tags.tag2
其中
*:表示将输入消息的各个字段都SELECT过来
'mymetric' AS name: 插入一个名为name, 值为'mymetric'的字段,方便后面用来作为metric
'name' AS _TSDB_META.global_metric:将前面生成的name字段,用来作为全局的metric
'ts' AS _TSDB_META.point_time:将数组中名为ts的字段,作为各个点的时间戳
'deviceid' AS _TSDB_META.global_tags.tag1:将消息的deviceid字段,作为一个全局的tag
整个对象(多度量默认域)
适合将某个JSON对象下面的每个数值类型字段分别写到TSDB,字段名为metric,域为默认域(value), 值为数据。
该方式需要在结果消息中包含一个名为"_TSDB_META_v2"的字段,描述以何种方式写TSDB
- metric_nodes: 待处理对象的JSON路径
- ts: 时间戳的路径
- tags: 一个或者多个tag的路径
- 时间戳格式(可选),如果时间戳为非标准格式,可以指定时间戳的格式,方便系统识别时间戳。在_TSDB_META_v2.time_format字段中指定时间戳格式,如:"yy-MM-dd HH:mm:ss"。
举例
结果消息的格式应该类似如下:
{
"reported": {
"memoryFree": 32,
"cpuIdle": 87.3
},
"sys_time": 1532488110931,
"deviceName": "myDeviceName",
"_TSDB_META_v2": {
"metric_nodes": {
"node1": "reported"
},
"ts": "sys_time",
"tags": {
"tag01": "deviceName"
}
}
}
如果设备上传的消息格式如下:
{
"reported": {
"memoryFree": 32,
"cpuIdle": 87.3
}
}
如果需要把reported对象下面的每个数值类型的属性(字符类型会被忽略)写入TSDB,可以通过如下SQL将消息变换成合法的格式:
*,
CURRENT_TIMESTAMP AS sys_time,
'myDeviceName' AS deviceName,
'reported' AS _TSDB_META_v2.metric_nodes.node1,
'sys_time' AS _TSDB_META_v2.ts,
'deviceName' AS _TSDB_META_v2.tags.tag01
其中,CURRENT_TIMESTAMP AS sys_time 这一句是将系统当前时间作为sys_time字段,方便后面作为写TSDB的时间戳。如果一个消息中,有多个对象需要写入,可以增加类型如下语句:
'another_object' AS _TSDB_META_v2.metric_nodes.node2,
'another_object2' AS _TSDB_META_v2.metric_nodes.node3
...
固定结构(多域写入)
多域写入方便将多个数据,当作完整的一组数据写入TSDB,好比MySql中一行中的多个字段,读取的时候可以一起返回。典型的例子有经纬度数据。该方式适合JSON格式相对固定的消息,并且可以精准控制哪些字段存TSDB。
该方式需要在结果消息中包含一个名为"_TSDB_META_v3"的字段,描述以何种方式写TSDB,指明如下几个参数:
- ts: 时间戳路径
- metric: 度量的路径
- tags: 一个或者多个tag的路径
- fields: 每个域在消息中的路径
举例
结果消息的格式应该类似如下:
{
"ts": 1531929753,
"carid": "car0001",
"gps": {
"longitude": 121.34,
"latitude": 30.12
},
"bin":"010aff",
"metric": "carlocation",
"_TSDB_META_v3": {
"ts": "ts",
"metric": "metric",
"tags": {
"tag1": "carid"
},
"fields": {
"field1": "gps.longitude",
"field2": "gps.latitude",
"hexAsBytesField0": "bin"
}
}
}
如果设备上传的消息格式为:
{
"ts":1531929753,
"carid":"car0001",
"gps": {
"longitude": 121.34,
"latitude": 30.12
},
"bin":"010aff"
}
需要把gps下面的longitude和latitude分别当作两个域,写入到名为carlocation的metric中去;把bin字段的HEX String表示的二进制数据,以bytes的方式写TSDB,则可以通过如下SQL语句进行变换:
*,
'carlocation' AS metric,
'ts' AS _TSDB_META_v3.ts,
'metric' AS _TSDB_META_v3.metric,
'carid' AS _TSDB_META_v3.tags.tag1,
'gps.longitude' AS _TSDB_META_v3.fields.field1,
'gps.latitude' AS _TSDB_META_v3.fields.field2,
'bin' AS _TSDB_META_v3.fields.hexAsBytesField0
单点(非默认域)
如果多个域的数据分别上传,则需要通过该方式把一个点的数据写到指定域中去。
该方式需要在结果消息中包含一个名为"_TSDB_META_v4"的字段,描述以何种方式写TSDB,
指明如下几个参数:
- ts: 时间戳
- metric: 度量名称
- field: 域的名字
- value: 值
- tags: 一个或者多个tag
举例
结果消息的格式应该类似如下:
{
"_TSDB_META_v4": {
"metric": "DeviceTemperature",
"field": "temperature",
"value": 34.6,
"timestamp": 1532483765,
"tags": {
"deviceid": "IMEI201807_002"
}
}
}
说明,_TSDB_META_v4对象与其他格式不同,他不引用外部的路径,而是最终的数据。 假如设备上传的消息格式如下:
{
"ts":1532483765,
"deviceid":"IMEI201807_002",
"temperature": 34.6
}
需要把temperate的数值写入到名为temperature的域,而不是默认的value域,则可以通过如下的SQL进行变换:
'DeviceTemperature' AS _TSDB_META_v4.metric,
'temperature' AS _TSDB_META_v4.field,
temperature AS `_TSDB_META_v4.value`,
ts AS `_TSDB_META_v4.timestamp`,
deviceid AS _TSDB_META_v4.tags.deviceid
不固定结构(多域写入)
对于像物影子的消息,每次上报的属性名称和个数不固定,但希望将某个对象下面所有的字段都以多域的方式写TSDB。Bool类型字段默认转化成0和1存储,可选地开启存储字符串类型字段。
该方式需要在结果消息中包含一个名为"_TSDB_META_v5"的字段,描述以何种方式写TSDB,指明如下几个参数:
- ts: 时间戳
- metric: 度量名
- tags: 一个或者多个tag
- fields: 该对象下面的每个子属性分别作为不同的域,写TSDB
- saveStringFields: 可选bool字段,指明是否将fields下面的字符串属性也存TSDB,默认为false
举例
结果消息的格式应该类似如下:
{
"_TSDB_META_v5": {
"metric": "mymetric",
"ts": 1535865771,
"fields": {
"temperature":25.6,
"status":true,
"pressure":"high"
},
"tags": {
"deviceName": "device001",
"model": "2018"
},
"saveStringFields": true
}
}
假设规则引擎订阅物管理主题$baidu/iot/shadow/+/update,收到类似如下的消息:
{
"reported": {
"temperature":25.6,
"status":true,
"pressure":"high"
}
}
以系统当前时间作为时间戳,从输入主题中提取影子名作为tag,则可以通过如下的SQL进行变换:
reported AS _TSDB_META_v5.fields, CURRENT_TIMESTAMP AS _TSDB_META_v5.ts, true AS _TSDB_META_v5.saveStringFields, 'shadow' AS _TSDB_META_v5.metric, SUBSTRING(topic(), 19, position('/update' in topic()) - 19) as _TSDB_META_v5.tags.deviceName
关系型数据库(RDS)
写入关系型数据库需要首先在百度的RDS付费并购买相应的实例,以及建好相应的库和表。
注意:表的列名必须和IoT Hub中的json格式的payload的各个key一一对应,并且大小写敏感,否则无法写入。
具体RDS的操作请参考RDS相关的产品文档。
第一次配置使用RDS需要先配置RDS的实例,具体操作步骤如下:
-
选择数据目的地为关系性数据库,点击创建连接配置,如下图所示
-
配置中配置需要规则引擎写入的RDS信息,其中名称可以任意填写唯一即可,主要用来标识这个配置的名称
- 配置结束后,下次使用可以选择已经配置好的连接配置,然后选择这个配置对应的库下面的某张表即可。
现在写入RDS支持insert和upsert:
- insert:适合记录所有消息。比如一个传感器不断上传自己的状态,如果需要将所有数据都写入RDS就适合用insert。其中这里面每一行数据有一个自增的主键。具体可以参考控制台里的sql模版。
- upsert:适合更新某一个设备的属性。规则引擎如果发现一个新的主键,会insert,如果这个主键已经存在,则更新整行数据。具体适合一些需要反向查询的场景,比如我们有100个传感器,每个传感器有5个属性,我们需要定期按照某个属性的特征,反向查询传感器的ID,这种就比较合适用upsert。这种用法相当于把设备的影子按列进行展开。具体可以参考控制台里的sql模版。
insert应用举例
在这个示例中,设备会上报两个信息:压力("pressure")和温度("temperature"),我们将设备的数据转存到RDS,以便兼容我们现有的技术栈。
前提:
假定已经在RDS创建了一个名为“mysql57”的MYSQL类型实例,并在实例中创建账号("account")及数据库("test_db") 并在DB中创建Table("test_table")如下:
CREATE TABLE `test_table` (
`id` int(10) NOT NULL AUTO_INCREMENT,
`pressure` varchar(40) NOT NULL,
`temperature` double(10,2) NOT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8
步骤:
-
创建"连接配置"(物联网console → 平台通用功能 → 连接配置),其中一个连接配置对应一个DB。
-
创建规则引擎
- 订阅物接入(设备型)的"$baidu/iot/shadow/{devicename}/update/snapshot"主题
-
提取其中的压力("pressure")和温度("temperature")信息,构建"RDS_META"的格式
-
物接入(设备型)的"$baidu/iot/shadow/{devicename}/update/snapshot"主题输入如下:
{ "reported": { "temperature":25.6, "pressure":"high" } }
-
规则引擎SQL如下:
reported.temperature AS RDS_META.columns.temperature, reported.pressure AS RDS_META.columns.pressure
-
产出如下:
{ "RDS_META":{ "columns":{ "temperature":25.6, "pressure":"high" } } }
-
- 订阅物接入(设备型)的"$baidu/iot/shadow/{devicename}/update/snapshot"主题
-
添加“关系型数据库”目的地
upsert应用举例
设备上报数据写到RDS里,我们希望根据给定的key,优先选择UPDATE现有记录,没有找到匹配的记录时再进行INSERT操作。这样我们可以简单的通过属性查询到对应的设备,如Select * From table WHERE temperature > 50
在这种示例中,设备会上报4个字段:temperature(温度), pressure(压力), devicename(设备名称), customerid(用户ID),其中devicename(设备名称)和customerid(用户ID)作为key
前提:
我们在数据库("test_db")中创建Table("upsert_test")如下:
CREATE TABLE `upsert_test` (
`id` int(10) NOT NULL AUTO_INCREMENT,
`devicename` varchar(40) NOT NULL,
`customerid` int(10) NOT NULL,
`temperature` double(10,2) NOT NULL,
`pressure` varchar(40) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `unique` (`devicename`,`customerid`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=2944 DEFAULT CHARSET=utf8
其中`devicename` + `customerid`做为一个组合UNIQUE KEY。
步骤:
-
创建"连接配置"(物联网console → 平台通用功能 → 连接配置)
因为一个"连接配置"对应一个DB。我们可以简单复用上述INSERT模式中创建的连接配置。
-
创建规则引擎
-
订阅物接入(设备型)的"$baidu/iot/shadow/{devicename}/update/snapshot"主题
-
构建"_RDS_META_"的UPSERT格式
-
物接入(设备型)的"$baidu/iot/shadow/{devicename}/update/snapshot"主题输入如下:
{ "reported": { "temperature":25.6, "pressure":"high", "devicename":"mydevice001", "customerid":2 } }
-
规则引擎SQL如下:重点在于作为key的`devicename`,`customerid`字段须在Table中建立为一个组合UNIQUE KEY
reported.temperature AS RDS_META.columns.temperature, reported.pressure AS RDS_META.columns.pressure, reported.devicename AS RDS_META.keys.devicename, reported.customerid AS RDS_META.keys.customerid
-
产出如下:
{ "RDS_META":{ "columns":{ "temperature":25.6, "pressure":"high" }, "keys":{ "devicename":"mydevice001", "customerid":2 } } }
-
-
-
添加“关系型数据库”目的地
物接入(IoT Hub)
将结果消息转发到另外的物接入主题中去,可以是到同一个实例、也可以到另外一个实例;可以是固定主题、也可以动态主题。
目的地主题可以是任何合法的主题,规则引擎有特权将消息转发到任何主题,但如果您需要订阅该主题,请确保您使用的设备有该主题的订阅权限,有必要时需要去物接入中配置权限。
固定主题
结果消息转发到预先定义的固定的主题,主题不能包含通配符:+#。可以选择不同的物接入实例,进而达到跨项目(实例)转发消息的能力。
动态主题
目的地主题并不固定,而是通过另外一个SQL语句从输入消息中SELECT出来,因此可以将不同的消息转发到不同的主题中去。该方式只能在同项目(实例)中转发,不能跨实例。
假如有如下两条消息:
{
"device": "device001"
}
{
"device": "device002"
}
并且主题选择器为:'destination/' || device,那么系统转发前,通过"SELECT 'destination/' || device"来计算真正的目的地主题,并且将上述两个消息分别转发到destination/device001, destination/device002两个不同的主题中去。其中"||"为字符串串联操作符。
BIN2JSON
视输入消息为二进制byte array,将其转化成Hex String,附上元信息,组合并且输出JSON格式消息。如Modbus RTU数据通过DTU透传到云端,可以通过该目的地将消息转化成物解析希望的格式(该格式可以在轮训设置右侧的上传格式中查看)。
举例
如果某个设备向MQTT主题/device/dtu/pub/topic发布了二进制消息0103020AFF,规则引擎将输出类似如下的JSON消息:
{
"msg":"0103020AFF",
"_meta_": {
"topic": "/device/dtu/pub/topic",
"clientid":"device01",
"clientip": "/180.2.3.4:51789",
"qos":0
}
}
- msg:为消息内容的Hex字符串表示
- _meta_.topic:为输入消息的MQTT主题
- _meta_.clientid:为输入消息的发布者ClientId
- _meta_.clientip:为输入消息的发布者的IP及端口
- _meta_.qos:为输入消息发布者发布所使用的QoS
百度消息服务(Kafka)
将结果消息转发到指定的Kafka主题。转发的key为null,默认的partitioner,因此,如果您的主题有多个分区,消息会均匀的发送到各个分区中去。
函数计算(CFC)
将结果消息作为输入调用指定的函数计算(CFC),并且可选地将CFC的返回转入另外一个MQTT主题。
函数计算仅接受JSON格式的输入,因此需要确保结果消息是JSON格式的。
规则字典
将结果消息写入到规则字典。规则引擎的结果消息需要包含如下JSON结构:
{
"_DICT_META": {
"k": "<the key>",
"v": <the value, of any type>
}
}
_DICT_META.k的值为规则字典的key,必须为字符串类型; _DICT_META.v的值为规则字典的value,类型可以是数值,可以是字符串,也可以是JSON对象;
举例1 假设规则字典存储的就是简单数值,即一个计数器,每当设备上报的alarm字段为1,则将字典中的计数器加一。
规则字典内容为:
key | value |
---|---|
car001 | 36 |
设备上报消息为:
{
"id": "car001",
"alarm": 1
}
SQL如下:
SELECT lookup_num(id) + 1 AS _DICT_META.v, id AS _DICT_META.k WHERE alarm = 1
产出如下:
{
"_DICT_META:{
"k": "car001",
"v": 37
}
}
举例2 假设规则字典存储是一个JSON格式,包含一个计数器和一个告警状态,每当告警状态从0变为1时,则将字典中的计数器加一。
规则字典内容为:
key | value |
---|---|
car001 | {"cnt":36, "alarm":0} |
设备上报消息为:
{
"id": "car001",
"alarm": 1
}
SQL如下:
SELECT CASE alarm WHEN 1 THEN lookup_num(id || '.cnt') + 1 ELSE lookup_num(id || '.cnt') END AS _DICT_META.v.cnt, alarm AS _DICT_META.v.alarm, id AS _DICT_META.k WHERE alarm <> lookup_num(id || '.alarm')
产出如下:
{
"_DICT_META:{
"k": "car001",
"v": {
"cnt": 37,
"alarm": 1
}
}
}
百度对象存储(BOS)
将每个结果消息以追加(append)的方式写到百度对象存储(BOS)文件中去,并且以日为单位生成不同的文件。
文件名的格式如下:
<uuid>_yyyy-MM-dd_N.ext
其中uuid为一个随机id; yyyy-MM-dd为当前日期,每天会生成一个新的文件;N为一个0开始的编号,每天从0开始,当一个文件的大小超过5GB时,生成新的文件,编号加一。
短信(SMS)
如果需要将结果消息以短信的方式发送,则需要选择一个短信接收人。
短信接收人是一个组合概念,在平台通用功能下面的短信接收人子菜单中创建。
短信仅接受JSON格式的输入,并且经过SQL变换后的结果消息需要与短信接收人中的短信模板匹配。
如果短信模板中有var1, var2两个变量,那么结果短信的JSON需要,并且仅需要包含var1, var2两个key。
举例
假如短信模板为:
物联网设备${devid}的温度值过高,最新值为${temp}
则结果消息只含有device, 和temp两个key,如:
{
"devid": "device01",
"temp": 34.5
}
如果设备输入的消息为:
{
"deviceid":"device01",
"temperature": 34.5
}
则可以通过如下SQL进行变换,使之符合SMS的格式要求:
deviceid AS devid, temperature AS temp
规则引擎(RULE)
规则引擎的消息可以转到不同的规则中去。
固定的规则引擎
结果消息转发到预先创建好的规则。
动态规则引擎
目的地规则并不固定,而是通过另外一个SQL语句从输入消息中SELECT出来,因此可以将不同的消息转发到不同的规则中去。
假如有如下一条消息:
{
"ruleName":"testRule1",
"temperature": 34.5
}
并且动态规则引擎为:ruleName,那么系统转发前,通过"SELECT ruleName"来计算真正的目的地规则,并且将上述消息转发到规则testRule1中去。