后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载导读本文围绕 EMQX 开源仓库中的emqx_bridge_timescale应用系统讲解 EMQX 如何通过数据桥接Bridge将 MQTT 消息批量写入时序数据库 TimescaleDB。由于 TimescaleDB 本质上是 PostgreSQL 的时序扩展EMQX 在实现上直接复用了 PostgreSQL 连接器与动作Action的整套配置体系本文将从模块结构、配置项、SQL 模板到 HTTP 管理 API 全链路展开帮助读者掌握创建连接器、编写规则 SQL、配置批量写入参数以及通过 API 管理桥接的完整方法。TimescaleDB 与 EMQX 数据集成背景根据 apps/emqx_bridge_timescale/README.md 的描述TimescaleDB 是一款面向时序数据设计的开源数据库其核心特性包括基于 PostgreSQL 构建TimescaleDB 不是独立数据库而是以 PostgreSQL 扩展extension的形式打包发布因此天然继承 PostgreSQL 的全部能力自动分区支持按时间维度time和空间维度partitioning key自动分区解决时序数据量持续增长下的存储与查询性能问题完整 SQL 支持用户可以直接使用标准 SQL 进行数据写入与聚合查询无需学习专用查询语言。而emqx_bridge_timescale这个应用的作用正是连接 EMQX 与 TimescaleDB用户可以创建一条规则Rule借助 EMQX 规则引擎EMQX Rules将海量 IoT 数据便捷地写入 TimescaleDB。在 EMQX 5.x 的桥接体系Bridge v2中桥接被拆分为**连接器Connector与动作Action**两个层次Connector 负责建立并管理到数据库的连接资源Action 负责在规则触发时执行具体的写入 SQL。为什么 TimescaleDB 桥接直接复用 PostgreSQL 实现从源码结构看TimescaleDB 桥接的整套 Schema 几乎全部委托给 PostgreSQL 相关模块这是本文最值得关注的设计点。在 apps/emqx_bridge_timescale/src/emqx_bridge_timescale.erl 中可以看到清晰的委托关系fields(config_connector)直接返回emqx_postgresql_connector_schema:fields(config_connector)即连接器配置完全复用 PostgreSQL 连接器fields(post)、fields(action)、fields(put_bridge_v2)等均委托给emqx_bridge_pgsql:fields(...)连接器示例connector_examples/1与桥接示例bridge_v2_examples/1也分别取自 PostgreSQL 连接器 Schema 与 PostgreSQL 桥接的values/1。更进一步在 apps/emqx_bridge_timescale/src/emqx_bridge_timescale_connector_info.erl 中type_name() - timescale、bridge_types() - [timescale]定义桥接类型标识resource_callback_module() - emqx_postgresql明确底层资源回调直接复用 PostgreSQL 驱动即实际建立数据库连接、执行 SQL 的模块与 PostgreSQL 桥接完全一致config_schema()引用emqx_bridge_timescale模块的config_connector字段。动作注册方面apps/emqx_bridge_timescale/src/emqx_bridge_timescale_action_info.erl 声明了action_type_name() - timescale、connector_type_name() - timescale与schema_module() - emqx_bridge_timescale。这两个 info 模块通过 apps/emqx_bridge_timescale/mix.exs 中的emqx_action_info_modules与emqx_connector_info_modules应用环境变量注册到 EMQX使系统能够在启动时发现并加载该桥接类型。因此可以总结出该模块的核心实现事实只要目标 PostgreSQL 实例安装了 TimescaleDB 扩展EMQX 的 PostgreSQL 数据通道即可无缝服务时序场景无需为 TimescaleDB 单独实现连接驱动这大幅降低了维护成本。连接器Connector配置项详解TimescaleDB 连接器的配置字段定义在 apps/emqx_postgresql/src/schema/emqx_postgresql_connector_schema.erl 的fields(connection_fields)中主要参数如下配置项类型/默认值说明server字符串形如127.0.0.1:5432PostgreSQL/TimescaleDB 服务地址默认端口定义在 apps/emqx_postgresql/include/emqx_postgresql.hrl 中为5432?PGSQL_DEFAULT_PORTdatabase字符串目标数据库名称示例值为emqx_datausername字符串必填数据库用户名示例值为postgrespassword字符串数据库密码示例值为publicpool_size整数默认8连接池大小application_name字符串默认emqx连接的应用名会通过validate_application_name/1校验disable_prepared_statements布尔是否禁用预处理语句ssl映射TLS 配置包含enable默认 false、verify、versions默认tlsv1.3/tlsv1.2、depth默认 10、ciphers、secure_renegotiate、reuse_sessions、hibernate_after等除连接字段外连接器还叠加了emqx_connector_schema:common_fields()如enable与资源选项resource_opts连接池、超时等资源级控制。一个典型的连接器配置示例如下取自 emqx_postgresql_connector_schema.erl 的示例值connectors { timescale my_timescale_connector { server 127.0.0.1:5432 database emqx_data username postgres password public pool_size 8 application_name emqx ssl { enable false } enable true } }动作Action与 SQL 模板TimescaleDB 动作的配置在 apps/emqx_bridge_pgsql/src/emqx_bridge_pgsql.erl 中定义动作核心参数action_parameters只有一个字段sql类型为 SQL 模板字符串emqx_schema:template()用于描述将消息写入哪张表、哪些列。其默认 SQL 为insert into t_mqtt_msg(msgid, topic, qos, payload, arrived) values (${id}, ${topic}, ${qos}, ${payload}, TO_TIMESTAMP((${timestamp} :: bigint)/1000))该模板展示了两个关键实践占位符绑定规则上下文${id}、${topic}、${qos}、${payload}、${timestamp}是 EMQX 规则引擎暴露的 SQL 占位符运行时会被消息对应的字段值替换时间戳转换TO_TIMESTAMP((${timestamp} :: bigint)/1000)将毫秒级时间戳转换为 TimescaleDB 可存储的时间类型充分利用时序数据库按时间分区的能力。动作层还包含resource_opts资源选项其默认值在fields(action_resource_opts)中定义batch_size默认100、batch_time默认100ms即开启批量写入——规则触发后最多积攒 100 条记录或等待 100ms 便批量刷入数据库。此外在 emqx_bridge_pgsql.erl 的示例值中还可看到inflight_window默认 100、max_buffer_bytes默认256MB、request_ttl默认45s、worker_pool_size默认 16等生产级调优项。这些批量参数对时序数据这种高频小消息场景尤为重要合理调大batch_size/batch_time可显著降低数据库写入压力。动作配置示例取自values(parameters)PostgreSQLType为timescale时生成bridges { timescale my_timescale_action { type timescale enable true connector my_timescale_connector resource_opts { batch_size 100 batch_time 100ms inflight_window 100 max_buffer_bytes 256MB request_ttl 45s worker_pool_size 16 } parameters { sql INSERT INTO client_events(clientid, event, created_at) VALUES (${clientid}, ${event}, TO_TIMESTAMP((${timestamp} :: bigint))) } } }实战接入流程结合上述配置将 IoT 数据写入 TimescaleDB 的标准流程如下准备数据库在 PostgreSQL 实例中启用 TimescaleDB 扩展CREATE EXTENSION timescaledb;创建目标数据库与数据表如t_mqtt_msg并按时间维度建立分区hypertable创建连接器通过 Dashboard 的“数据集成”页面或 HTTP API 创建timescale类型连接器填写上文连接器配置项创建动作在连接器下创建timescale类型动作编写 SQL 模板参考默认模板可自定义表名与占位符映射并按需调整批量参数编写规则创建一条 EMQX 规则选择消息发布message publish等事件作为触发源将动作挂载到规则上实现“MQTT 消息 → 规则 → SQL → TimescaleDB”的完整链路。桥接管理的 HTTP API如 apps/emqx_bridge_timescale/README.md 所述EMQX 为桥接管理提供了一组 HTTP API覆盖创建桥接、更新桥接、查询桥接、停止/重启桥接、列出桥接等操作。这些 API 由桥接管理模块统一提供emqx_bridge_timescale通过fields(post_bridge_v2)、fields(get_bridge_v2)、fields(put_bridge_v2)分别对应创建、查询、更新接口的请求/响应结构连接器侧则由emqx_connector_schema:api_ref/3见 emqx_bridge_timescale_connector_info.erl生成POST/PUT/GET .../connectors/{type}系列接口。典型用法包括创建/更新后调用查询接口确认连接状态status、node_status是否connected写入异常或数据库维护后通过停止/重启接口快速恢复数据通道列出所有桥接以便统一巡检资源健康度。小结EMQX 的 TimescaleDB 桥接是一个“复用 PostgreSQL 通道、叠加时序能力”的典型实现连接驱动完全复用emqx_postgresqlSchema 复用 PostgreSQL 连接器与桥接定义桥接类型标识统一为timescale。开发者只需保证目标库启用 TimescaleDB 扩展即可用标准 SQL 模板配合规则引擎将 MQTT 数据批量写入时序表并通过批量参数与 HTTP API 实现生产级的写入性能与运维管理。相关实现可进一步阅读 emqx_bridge_timescale.erl、emqx_bridge_pgsql.erl 与 emqx_postgresql_connector_schema.erl贡献指南见仓库根目录的 CONTRIBUTING.md。赞分享后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载相关推荐Glances 集成 TimescaleDB基于 PostgreSQL 的时序数据库监控数据导出实战Glances 集成 TimescaleDB基于 PostgreSQL 的时序数据库监控数据导出实战 Glances 内置了面向 TimescaleDB 的导指标监控监控大盘CLI告警MCP 服务革命性多语言社交媒体AI模型TwHIN-BERT-large完全指南革命性多语言社交媒体AI模型TwHIN BERT large完全指南 TwHIN BERT large 是一个革命性的多语言社交媒体AI模型专门为理解和处理EMQX与PostgreSQL集成关系型数据库数据桥接全指南EMQX与PostgreSQL集成关系型数据库数据桥接全指南 引言物联网数据流的关系型数据库落地挑战 在工业物联网IIoT场景中设备产生的实时数据需要后端物联网消息队列通信上一篇SGLang项目单GPU运行多模型实例的技术方案解析下一篇解决SGLang项目在Kubernetes中部署GPU容器的8大痛点问题创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考