数据转发到函数计算服务CFC

更新时间：2025-08-21

一、简介

在本教程中，您将学习到如何将 IoT Core 的设备消息通过「规则引擎」转发至「函数计算 CFC」，并通过「函数计算 CFC」进行用户自定义的数据处理。

二、应用场景描述

规则引擎可以做到海量数据的过滤、变型和转发，但当您对设备数据处理有着更负复杂，更个性化的处理需求时，仅仅使用规则引擎可能无法满足全部需求，但可以结合「函数计算 CFC」来满足相关处理需求。通过将「规则引擎」与「函数计算 CFC」的组合，数据的处理将会变得更加灵活多样。主要应用场景包括：

进行复杂场景的数据处理。
存储设备日志信息。
转发设备消息到用户自定义服务。

三、示例场景描述

为了带您深入了解「规则引擎」与「函数计算 CFC」的组合使用方式，将采用一个简单的示例：

规则引擎将设备产生的系统日志转发到 CFC。
CFC 对接受到的日志消息进行解码，并将消息内容存储到对象存储 BOS 中。

操作步骤如下：

创建对象存储 BOS 实例。
创建函数计算 CFC 实例，并配置 BOS 为日志存储目的地。
创建 IoT Core 实例，创建转发规则。
验证设备日志转发存储功能。

四、操作步骤

4.1 创建 BOS 实例

具体方式请参考《对象存储 BOS 创建Bucket》

登录对象存储 BOS 管理控制台。点击左侧导航栏 " Bucket 列表"，在 Bucket 列表中找到并点击 + 按钮（ “ 新建 Bucket ” 按钮），在弹出框中按照提示创建 Bucket。

点击「确定」后，该 Bucket 创建完成。

4.2 创建 CFC 实例并配置目的地

选择创建空白函数。

在「函数配置」中，选择「对象存储 BOS」作为日志存储，日志存储路径自行定义。

「触发器」选择无。

因为规则引擎会将消息内容以 Base64 编码后发送给CFC，详情请参考《转发到函数 CFC》。所以，当我们需要在 CFC 中获取规则引擎传输的数据时，需要进行如下两个步骤：

1.将event['message']中的消息进行base64解码。

2.根据 Protocol Buffers 的描述文件，将解码后的数据转换为对象。

在本教程中，函数计算 CFC 函数运行时采用 Python3.6，其功能为：将设备数据解码并打印 log。

Python 函数程序参考代码如下：

Plain Text

1import base64
2import time
3from google.protobuf import json_format
4import json
5import MqttLogConstants_pb2 as MqttLog
6from TrafficLogEntry_pb2 import TrafficLogEntries
7import logging
8logging.basicConfig()
9logger = logging.getLogger()
10logger.setLevel(logging.INFO)
11
12# 解码 byte 数据，转化为 log 对象
13def parse_log_entries(bytes):
14    entries = TrafficLogEntries()
15    entries.ParseFromString(bytes)
16    return entries
17
18# CFC 函数入口
19def handler(event, context):
20    # 进行base64解码
21    logBytes=base64.b64decode(event['message'])
22    # 解码 byte 数据，转化为 log 对象
23    logEntries = parse_log_entries(logBytes)
24    for entry in logEntries.entries:
25        if entry.code.startswith('MQT'):
26            entryJson = json.loads(json_format.MessageToJson(entry))
27            details = entryJson['details']
28            newDetails = {}
29            for keyNum in details:
30                detail = MqttLog.LogDetailKey.Name(int(keyNum))
31                newDetails[detail] = details[keyNum]
32            entryJson['details'] = newDetails
33            # 将解码结果作为日志打印
34            logger.info(entryJson)
35    return None