使用托管数据库部署 Coreflux MQTT 代理MQTT 代理 通过发布-订阅消息模式连接物联网设备和应用程序，使其成为现代 物联网 基础设施的重要组成部分。Coreflux 是一个 低代码 MQTT 代理，增加了实时数据处理和转换功能，让你可以直接与 DigitalOcean 托管数据库（包括 MongoDB、PostgreSQL、MySQL 和 OpenSearch）集成，而无需编写自定义集成代码。你将学到什么： 本教程将引导你部署一个完整的物联网数据管道——从在 DigitalOcean 上设置托管数据库集群和 Coreflux MQTT 代理，到配置安全的 VPC 网络、使用 Coreflux 的物联网语言 (LoT) 构建数据转换模型，以及自动将处理后的物联网数据存储到你选择的数据库中。最终你将获得一个可用于生产环境的设置，能够处理物联网应用的实时消息传递和持久存储。关键要点在深入了解分步部署过程之前，以下是你将学到的关键点：在 DigitalOcean 上部署托管数据库集群（PostgreSQL、MongoDB、MySQL 或 OpenSearch），用于可扩展的物联网数据存储。使用 Marketplace 镜像或 Docker 在 DigitalOcean Droplet （DigitalOcean的VPC）上设置 Coreflux MQTT 代理。创建安全的 VPC 网络以连接你的 MQTT 代理和数据库，无需公开暴露。使用 Coreflux 的物联网语言 (LoT) 构建实时数据管道，实现低代码物联网自动化。自动转换和存储物联网数据，从 MQTT 主题到数据库表、集合或索引。验证端到端数据流，从模拟传感器通过转换模型到数据库存储。本教程为需要实时消息传递结合持久数据存储以及搜索、分析或关系查询等高级功能的物联网应用提供了一个可用于生产环境的基础。你将构建什么在本教程结束时，你将得到：一个用于可扩展存储的托管数据库集群（PostgreSQL、MongoDB、MySQL 或 OpenSearch）一台运行 Coreflux MQTT 代理的 DigitalOcean Droplet一个用于安全 物联网通信的虚拟私有云 (VPC) 网络使用 LoT Notebook 扩展进行的实时数据模拟低代码数据转换模型和数据库集成路由用于 物联网自动化 的完整 数据集成与转换 管道Coreflux 与 DigitalOcean 合作Coreflux 通过物联网语言编程语言在 DigitalOcean 云平台上提供轻量级 MQTT 代理和数据管道工具，以实现高效的物联网通信。什么是 MQTT？MQTT（消息队列遥测传输）是一种轻量级的、发布-订阅网络协议，在物联网生态系统中被广泛采用。专为受限设备和低带宽、高延迟或不稳定的网络设计，MQTT 能够在带宽受限的环境中实现高效、实时的消息传递。关于 CorefluxCoreflux 提供了一个轻量级 MQTT 代理，以促进物联网设备与应用程序之间的高效、实时通信，包括每个用例所必需的实时数据转换功能。为可扩展性和可靠性而构建，Coreflux 专为低延迟和高吞吐量至关重要的环境量身定制。无论你是构建一个小型物联网项目还是部署工业监控系统，Coreflux 都能处理设备之间的消息路由和数据流。在 DigitalOcean 云平台上使用 Coreflux，你将获得：数据处理： 在你的数据所在之处集中处理你的数据处理需求，确保实时数据处理。数据集成： 轻松与其他 DigitalOcean 服务（如托管数据库 PostgreSQL、MongoDB、MySQL 或 OpenSearch）集成，确保为你的所有数据需求提供一个单一且简单的生态系统。可扩展性： 轻松处理不断增长的数据和设备数量，而不会影响性能。可靠性： 确保在所有连接的设备之间进行一致且可靠的消息传递。准备工作在开始本 MQTT 代理 部署教程之前，你需要：一个 DigitalOcean 帐户，可在DigitalOcean.com注册，支持绑定信用卡、支付宝或数字货币了解 MQTT 协议概念和 物联网 架构Visual Studio Code（用于 LoT Notebook 扩展）预计时间： 本教程大约需要 30-45 分钟完成，具体取决于数据库配置时间（通常每个数据库集群需要 1-5 分钟）。步骤 1 — 为物联网自动化创建网络基础设施为安全的 MQTT 通信创建 VPC 网络首先，你将创建一个虚拟私有云 (VPC)，以确保你的 物联网 服务和 MQTT 代理 之间的安全通信，无需公开访问。登录你的 DigitalOcean 控制面板从左侧导航栏进入 网络 → VPC点击 创建 VPC 网络为物联网自动化配置你的 VPC：名称：coreflux-integrations-vpc（或你的 VPC 名称）数据中心区域：选择法兰克福（或你首选的区域）IP 范围：使用默认值或根据需要配置描述：为你的 MQTT 代理和数据库 网络添加有意义的描述点击 创建 VPC 网络VPC 将为你所有的物联网资源提供隔离的网络，确保 Coreflux MQTT 代理 和 托管数据库 之间的安全通信。有关 VPC 配置的更多详细信息，请参阅我们关于创建 VPC 网络的教程。步骤 2 — 为可扩展存储设置托管数据库根据你的物联网应用需求，选择以下数据库选项之一：PostgreSQL：适用于需要关系查询、ACID 合规性和复杂关系的结构化数据MySQL：适用于结构化工作负载和具有强一致性及广泛工具支持的事务性查询MongoDB：适用于具有可变模式的灵活文档存储和快速开发OpenSearch：适用于高级搜索、分析、日志分析和时间序列数据可视化设置 PostgreSQL 托管数据库当你的物联网工作负载需要关系模式、强一致性 和 高级 SQL 分析，并由自动备份、监控和维护支持时，DigitalOcean 上的托管 PostgreSQL 是一个很好的选择。从 DigitalOcean 控制面板，导航到 数据库点击 创建数据库集群为物联网自动化配置你的 PostgreSQL 集群：数据库引擎：选择 PostgreSQL版本：选择最新的稳定版本数据中心区域：选择法兰克福（与你的 VPC 相同）VPC 网络：选择你创建的 coreflux-integrations-vpc数据库集群名称：postgresql-coreflux-test项目：选择你的目标项目根据你的 物联网 需求选择你的计划：对于开发：基础 计划，1 GB RAM对于生产：通用型 或更高，用于可扩展存储点击 创建数据库集群托管数据库 创建过程通常需要 1-5 分钟。完成后，你将被重定向到数据库概览页面，在那里你可以查看连接详细信息并执行管理操作。为 MQTT 代理集成配置 PostgreSQL 数据库访问系统将提示你进行入门步骤，显示你的连接详细信息，你可以配置入站访问规则（建议限制为你的 IP 和仅 VPC）。点击 开始使用 来配置你的 PostgreSQL 数据库（可选操作）限制入站连接：添加你本地计算机的 IP 以进行管理访问droplet 将通过 VPC 网络自动获得允许对于连接详细信息，你将看到两个选项 - 公共网络和 VPC 网络。第一个用于像 DBeaver 这样的工具进行外部访问，而第二个将由 Coreflux 服务用于访问数据库。记下提供的连接详细信息，包括公共访问和 VPC 访问（每种都有不同的详细信息）：主机：你的数据库主机名用户：默认管理员用户密码：自动生成的安全密码数据库：身份验证数据库名称测试 PostgreSQL 数据库连接你可以使用提供的连接参数，使用公共访问凭证通过 DBeaver 测试 PostgreSQL 连接：创建 PostgreSQL 应用程序数据库和用户（可选）为了更好的安全性和组织性，为你的 物联网自动化 应用程序创建一个专用用户和数据库。这也可以通过 DBeaver 或 CLI 完成，但 DigitalOcean 提供了一种用户友好的方法：转到你的 托管数据库 集群中的 用户与数据库 选项卡创建用户：用户名：coreflux-broker-client密码：自动生成创建数据库：数据库名称：coreflux-broker-data注意： 你可能需要更改数据库内的用户权限，以便能够创建表、插入和选择数据。对于 PostgreSQL，使用 GRANT CREATE, INSERT, SELECT ON DATABASE coreflux-broker-data TO coreflux-broker-client; 授予必要的权限。对于 MySQL，使用 GRANT CREATE, INSERT, SELECT ON coreflux-broker-data.* TO 'coreflux-broker-client'@'%';。设置 MySQL 托管数据库当你想要熟悉的 SQL、广泛的生态系统支持以及处理备份、更新和监控的完全托管服务时，DigitalOcean 上的托管 MySQL 是结构化、事务性物联网数据的理想选择。从 DigitalOcean 控制面板，导航到 数据库点击 创建数据库集群为物联网自动化配置你的 MySQL 集群：数据库引擎：选择 MySQL版本：选择最新的稳定版本数据中心区域：选择法兰克福（与你的 VPC 相同）VPC 网络：选择你创建的 coreflux-integrations-vpc数据库集群名称：mysql-coreflux-test项目：选择你的目标项目根据你的 物联网 需求选择你的计划：对于开发：基础 计划，1 GB RAM对于生产：通用型 或更高，用于可扩展存储点击 创建数据库集群托管数据库 创建过程通常需要 1-5 分钟。完成后，你将被重定向到数据库概览页面，在那里你可以查看连接详细信息并执行管理操作。为 MQTT 代理集成配置 MySQL 数据库访问系统将提示你进行入门步骤，显示你的连接详细信息，你可以配置入站访问规则（建议限制为你的 IP 和仅 VPC）。点击 开始使用 来配置你的 MySQL 数据库（可选操作）限制入站连接：添加你本地计算机的 IP 以进行管理访问droplet 将通过 VPC 网络自动获得允许对于连接详细信息，你将看到两个选项 - 公共网络和 VPC 网络。第一个用于像 DBeaver 这样的工具进行外部访问，而第二个将由 Coreflux 服务用于访问数据库。记下提供的连接详细信息，包括公共访问和 VPC 访问（每种都有不同的详细信息）：主机：你的数据库主机名用户：默认管理员用户密码：自动生成的安全密码数据库：身份验证数据库名称测试 MySQL 数据库连接你可以使用提供的连接参数，使用公共访问凭证通过 DBeaver 测试 MySQL 连接。注意： 你可能需要更改 DBeaver 的驱动程序设置——设置 allowPublicKeyRetrieval = true。创建 MySQL 应用程序数据库和用户（可选）为了更好的安全性和组织性，为你的 物联网自动化 应用程序创建一个专用用户和数据库。这也可以通过 DBeaver 或 CLI 完成，但 DigitalOcean 提供了一种用户友好的方法：转到你的 托管数据库 集群中的 用户与数据库 选项卡创建用户：用户名：coreflux-broker-client密码：自动生成创建数据库：数据库名称：coreflux-broker-data设置 MongoDB 托管数据库托管 MongoDB 非常适合灵活或不断演变的物联网负载，让你能够存储异构的传感器文档，而无需严格模式，同时平台处理复制、备份和监控。从 DigitalOcean 控制面板，导航到 数据库点击 创建数据库集群为物联网自动化配置你的 MongoDB 集群：数据库引擎：选择 MongoDB版本：选择最新的稳定版本数据中心区域：选择法兰克福（与你的 VPC 相同）VPC 网络：选择你创建的 coreflux-integrations-vpc数据库集群名称：mongodb-coreflux-test项目：选择你的目标项目根据你的 物联网 需求选择你的计划：对于开发：基础 计划，1 GB RAM对于生产：通用型 或更高，用于可扩展存储点击 创建数据库集群托管数据库 创建过程通常需要 1-5 分钟。完成后，你将被重定向到数据库概览页面，在那里你可以查看连接详细信息并执行管理操作。为 MQTT 代理集成配置 MongoDB 数据库访问系统将提示你进行入门步骤，显示你的连接详细信息，你可以配置入站访问规则（建议限制为你的 IP 和仅 VPC）。点击 开始使用 来配置你的 MongoDB 数据库（可选）限制入站连接：添加你本地计算机的 IP 以进行管理访问droplet 将通过 VPC 网络自动获得允许对于连接详细信息，你将看到两个选项：公共网络和 VPC 网络。第一个用于像 MongoDB Compass 这样的工具进行外部访问，而第二个将由 Coreflux 服务用于访问数据库。记下提供的连接详细信息，包括公共访问和 VPC 访问（每种都有不同的详细信息）：主机：你的数据库主机名用户：默认管理员用户密码：自动生成的安全密码数据库：身份验证数据库名称测试 MongoDB 数据库连接你可以使用 MongoDB Compass 或提供的连接字符串，使用公共访问凭证测试 MongoDB 连接：mongodb://username:password@mongodb-host:27017/defaultauthdb?ssl=true创建 MongoDB 应用程序数据库和用户（可选）为了更好的安全性和组织性，为你的 物联网自动化 应用程序创建一个专用用户和数据库。这也可以通过 MongoDB Compass 或 CLI 完成，但 DigitalOcean 提供了一种用户友好的方法：转到你的 托管数据库 集群中的 用户与数据库 选项卡创建用户：用户名：coreflux-broker-client密码：自动生成创建数据库：数据库名称：coreflux-broker-data设置 OpenSearch 托管数据库托管 OpenSearch 专为高容量物联网数据的搜索、日志分析和时间序列仪表板而设计，该服务为你管理集群健康、扩展和索引存储。从 DigitalOcean 控制面板，导航到 数据库点击 创建数据库集群为物联网自动化配置你的 OpenSearch 集群：数据库引擎：选择 OpenSearch版本：选择最新的稳定版本数据中心区域：选择法兰克福（与你的 VPC 相同）VPC 网络：选择你创建的 coreflux-integrations-vpc数据库集群名称：opensearch-coreflux-test项目：选择你的目标项目根据你的 物联网 需求选择你的计划：对于开发：基础 计划，1 GB RAM对于生产：通用型 或更高，用于可扩展存储点击 创建数据库集群托管数据库 创建过程通常需要 1-5 分钟。完成后，你将被重定向到数据库概览页面，在那里你可以查看连接详细信息并执行管理操作。为 MQTT 代理集成配置 OpenSearch 数据库访问系统将提示你进行入门步骤，显示你的连接详细信息，你可以配置入站访问规则（建议限制为你的 IP 和仅 VPC）。点击 开始使用 来配置你的 OpenSearch 数据库（可选）限制入站连接：添加你本地计算机的 IP 以进行管理访问droplet 将通过 VPC 网络自动获得允许对于连接详细信息，你将看到两个选项：公共网络和 VPC 网络。第一个用于工具的外部访问，而第二个将由 Coreflux 服务用于访问数据库。你还将看到访问 OpenSearch 仪表板的 URL 和参数。记下提供的连接详细信息，包括公共访问和 VPC 访问（每种都有不同的详细信息）：主机：你的数据库主机名用户：默认管理员用户密码：自动生成的安全密码数据库：身份验证数据库名称测试 OpenSearch 数据库连接你可以使用提供的凭证通过 OpenSearch 仪表板测试 OpenSearch 连接：步骤 3 — 在 DigitalOcean Droplet 上部署 Coreflux MQTT 代理创建 DigitalOcean Droplet在你的 DigitalOcean 控制面板中导航到 Droplets点击 创建 Droplet为 MQTT 代理 部署配置你的 droplet：选择区域：法兰克福（与你的托管数据库相同）VPC 网络：选择 coreflux-integrations-vpc选择镜像：转到 Marketplace 选项卡搜索 “Coreflux” 并从 Marketplace 中选择 Coreflux为你的 物联网 工作负载选择大小：对于开发：基础 计划，2 GB 内存对于生产：基础 或 通用型 计划，4+ GB 内存以获得可扩展性能选择身份验证方法：SSH 密钥：推荐用于提高安全性可以使用 ssh-keygen 在本地创建密钥密码：备选方案最终确定详细信息：主机名：coreflux-test-broker项目：选择你的项目标签：为 DevOps 组织添加相关标签点击 创建 Droplet查看 Droplet 主页并等待其完成部署替代方案 - 在Docker镜像Droplet上使用Docker安装Coreflux MQTT代理采用与Coreflux Droplet相同的方法，选择Docker作为市场应用镜像。一旦你的droplet运行起来，通过已定义的认证方法或Droplet主页上提供的Web控制台，使用SSH连接到它：ssh root@your-droplet-ip使用Docker运行Coreflux MQTT代理：docker run -d \ --name coreflux \ -p 1883:1883 \ -p 1884:1884 \ -p 5000:5000 \ -p 443:443 \ coreflux/coreflux-mqtt-broker-t:1.6.3这个Docker命令：以分离模式运行容器 (-d)将容器命名为 coreflux暴露MQTT和Web界面所需的端口使用最新的Coreflux镜像验证MQTT代理是否在运行：docker ps你应该看到一个正在运行的容器：通过使用默认值连接到MQTT代理来验证部署你可以通过MQTT客户端（如MQTT Explorer）访问MQTT代理，以验证对代理的访问，无论采用何种部署方法。步骤4 — 为安全的物联网通信配置防火墙规则（可选）对于生产环境的物联网自动化部署，配置防火墙规则以限制访问：导航到网络 → 防火墙点击创建防火墙配置MQTT代理安全的入站规则：SSH：来自你IP的端口22MQTT：来自你的物联网应用程序源的端口1883带TLS的MQTT：用于安全的带TLS的MQTT的端口1884WebSocket：用于通过WebSocket的MQTT的端口5000带TLS的WebSocket：用于通过带TLS的WebSocket的MQTT的端口443将防火墙应用到你的droplet关于详细的防火墙配置，请参考DigitalOcean的防火墙快速入门教程。生产提示： 将MQTT端口1883限制在特定的源IP或VPC范围，并且对于外部设备连接，优先使用端口1884（带TLS的MQTT）。如果你需要额外的安全层，请考虑使用带有私有网络的DigitalOcean应用平台。步骤5 — 使用Coreflux的Language of Things设置物联网数据集成安装LoT Notebook扩展用于Visual Studio Code的LoT（Language of Things）Notebook扩展提供了一个集成的低代码开发环境，用于MQTT代理编程和物联网自动化。了解更多关于Coreflux的Language of Things (LoT)用于低代码物联网自动化的信息。打开Visual Studio Code转到扩展（Ctrl+Shift+X）搜索"LoT Notebooks"安装由Coreflux提供的LoT VSCode Notebooks扩展连接到你的MQTT代理配置与你的Coreflux MQTT代理的连接，当在顶部栏提示时或通过点击底部左侧栏的MQTT按钮时，使用默认凭据：用户：root密码：coreflux假设没有错误，你将在底部左侧栏看到与代理的MQTT连接状态。步骤6 — 通过Actions在MQTT代理中创建数据对于这个用例，我们将通过一个转换管道将原始数据集成到数据库中。然而，由于在演示中没有连接到任何MQTT设备，我们将利用LoT的能力，并使用一个Action来模拟设备数据。在LoT中，Action是一种可执行的逻辑，由特定事件触发，例如定时间隔、主题更新或其他操作或系统组件的显式调用。Actions允许与MQTT主题、内部变量和负载进行动态交互，促进复杂的物联网自动化工作流。因此，我们可以使用一个以定义的时间间隔在特定主题中生成数据的Action，然后由我们将在下面定义的管道的其余部分使用。你可以下载包含示例项目的github仓库。生成模拟物联网数据使用低代码的LoT（Language of Things）界面创建一个Action来生成模拟传感器数据：DEFINE ACTION RANDOMIZEMachineData ON EVERY 10 SECONDS DO PUBLISH TOPIC "raw_data/machine1" WITH RANDOM BETWEEN 0 AND 10 PUBLISH TOPIC "raw_data/station2" WITH RANDOM BETWEEN 0 AND 60在提供的Notebook中，你还有一个Action可以执行递增计数器来模拟数据，作为提供Action的替代方案。当你运行这个Action时，它将：自动部署到MQTT代理每10秒生成一次模拟的物联网传感器数据将实时数据发布到特定的MQTT主题在LoT Notebook界面中显示同步状态此状态显示LoT Notebook上的代码是否与代理中运行的代码不同，或者是否完全缺失步骤7 — 为实时处理创建数据转换模型使用Language of Things定义数据模型Coreflux中的模型用于转换、聚合和计算来自输入MQTT主题的值，并将结果发布到新主题。它们是创建适用于你多个数据源的UNS - 统一命名空间 - 的基础。因此，通过该模型，你可以定义原始物联网数据的结构与转换方式，适用于单个设备，也支持同时处理多个设备（借助通配符+实现）。模型还作为用于可扩展存储到托管数据库的关键数据模式。DEFINE MODEL MachineData WITH TOPIC "Simulator/Machine/+/Data" ADD "energy" WITH TOPIC "raw_data/+" AS TRIGGER ADD "energy_wh" WITH (energy * 1000) ADD "production_status" WITH (IF energy > 5 THEN "active" ELSE "inactive") ADD "production_count" WITH (IF production_status EQUALS "active" THEN (production_count + 1) ELSE 0) ADD "stoppage" WITH (IF production_status EQUALS "inactive" THEN 1 ELSE 0) ADD "maintenance_alert" WITH (IF energy > 50 THEN TRUE ELSE FALSE) ADD "timestamp" WITH TIMESTAMP "UTC"这个低代码模型：使用通配符+自动应用到所有机器通过乘以1000将能量转换为瓦时（energy_wh）根据能量阈值确定生产状态跟踪生产计数和停机事件向所有实时数据点添加时间戳从主题结构中提取机器ID将结构化数据发布到Simulator/Machine/Data主题（将+替换为每个匹配触发器/源数据格式的主题）由于我们使用Action生成了两个模拟传感器/机器，我们可以看到模型结构自动应用于两者，同时生成了一个json对象和各个单独的主题。步骤8 — 为可扩展存储设置数据库集成选择与你在步骤2中选择的数据库相匹配的数据库集成部分。PostgreSQL集成在本节中，你将学习如何将处理后的物联网数据存储到DigitalOcean上的PostgreSQL托管数据库中。要将处理后的物联网数据存储到PostgreSQL托管数据库中，你需要在Coreflux中定义一个Route。Route使用简单、低代码的配置指定数据如何从你的MQTT代理发送到你的PostgreSQL集群：DEFINE ROUTE PostgreSQL_Log WITH TYPE POSTGRESQL ADD SQL_CONFIG WITH SERVER "db-postgresql.db.onmyserver.com" WITH PORT 25060 WITH DATABASE "defaultdb" WITH USERNAME "doadmin" WITH PASSWORD "AVNS_pass" WITH USE_SSL TRUE WITH TRUST_SERVER_CERTIFICATE FALSE使用来自DigitalOcean的你自己的PostgreSQL连接详细信息替换，并在你的LoT Notebook中运行该Route。重要提示： 为了更好的安全性和更低的延迟，请使用VPC连接详细信息（而非公共连接）。VPC主机名和端口与公共连接字符串不同 - 请检查你的数据库集群的连接详细信息页面以获取这两个选项。为PostgreSQL数据库存储更新模型修改你的LoT模型以使用数据库路由进行可扩展存储，通过将此添加到模型的末尾：STORE IN "PostgreSQL_Log" WITH TABLE "MachineProductionData"此外，添加一个带有主题的参数，以便在你的托管数据库中为每个条目提供唯一标识符。DEFINE MODEL MachineData WITH TOPIC "Simulator/Machine/+/Data" ADD "energy" WITH TOPIC "raw_data/+" AS TRIGGER ADD "device_name" WITH REPLACE "+" WITH TOPIC POSITION 2 IN "+" ADD "energy_wh" WITH (energy * 1000) ADD "production_status" WITH (IF energy > 5 THEN "active" ELSE "inactive") ADD "production_count" WITH (IF production_status EQUALS "active" THEN (production_count + 1) ELSE 0) ADD "stoppage" WITH (IF production_status EQUALS "inactive" THEN 1 ELSE 0) ADD "maintenance_alert" WITH (IF energy > 50 THEN TRUE ELSE FALSE) ADD "timestamp" WITH TIMESTAMP "UTC" STORE IN "PostgreSQL_Log" WITH TABLE "MachineProductionData"部署此更新后的操作后，所有数据在更新时应自动存储在数据库中。MySQL集成MySQL是一种广泛使用的关系数据库管理系统，非常适合大规模存储和分析物联网数据。在本节中，你将学习如何将你的Coreflux MQTT代理连接到DigitalOcean上的托管MySQL数据库，以便你的实时设备数据能够安全可靠地持久化，用于分析、报告或与其他应用程序集成。要启用此集成，你必须在Coreflux的LoT（Language of Things）中定义一个Route，指示处理后的数据应该发送到哪里以及如何发送。下面是路由数据到MySQL数据库所需的低代码格式。请务必根据需要替换你自己的连接详细信息：DEFINE ROUTE MySQL_Log WITH TYPE MYSQL ADD SQL_CONFIG WITH SERVER "db-mysql.db.onmyserver.com" WITH PORT 25060 WITH DATABASE "defaultdb" WITH USERNAME "doadmin" WITH PASSWORD "AVNS_pass" WITH USE_SSL TRUE WITH TRUST_SERVER_CERTIFICATE FALSE使用来自DigitalOcean的你自己的MySQL连接详细信息替换，并在你的LoT Notebook中运行该Route。重要提示： 为了更好的安全性和更低的延迟，请使用VPC连接详细信息（而非公共连接）。如果你遇到连接问题，请验证TRUST_SERVER_CERTIFICATE是否已为你的MySQL版本正确设置 - 某些版本需要TRUE，而其他版本则使用FALSE。为MySQL数据库存储更新模型修改你的LoT模型以使用数据库路由进行可扩展存储，通过将此添加到模型的末尾：STORE IN "MySQL_Log" WITH TABLE "MachineProductionData"此外，添加一个带有主题的参数，以便在你的托管数据库中为每个条目提供唯一标识符。DEFINE MODEL MachineData WITH TOPIC "Simulator/Machine/+/Data" ADD "energy" WITH TOPIC "raw_data/+" AS TRIGGER ADD "device_name" WITH REPLACE "+" WITH TOPIC POSITION 2 IN "+" ADD "energy_wh" WITH (energy * 1000) ADD "production_status" WITH (IF energy > 5 THEN "active" ELSE "inactive") ADD "production_count" WITH (IF production_status EQUALS "active" THEN (production_count + 1) ELSE 0) ADD "stoppage" WITH (IF production_status EQUALS "inactive" THEN 1 ELSE 0) ADD "maintenance_alert" WITH (IF energy > 50 THEN TRUE ELSE FALSE) ADD "timestamp" WITH TIMESTAMP "UTC" STORE IN "MySQL_Log" WITH TABLE "MachineProductionData"部署此更新后的操作后，所有数据在更新时应自动存储在数据库中。MongoDB集成MongoDB是一种NoSQL数据库，非常适合存储和查询具有灵活模式的物联网数据。在本节中，你将学习如何将你的Coreflux MQTT代理连接到DigitalOcean上的托管MongoDB数据库，以便你的实时设备数据能够安全可靠地持久化，用于分析、报告或与其他应用程序集成。要启用此集成，你必须在Coreflux的LoT（Language of Things）中定义一个Route，指示处理后的数据应该发送到哪里以及如何发送。下面是路由数据到MongoDB数据库所需的低代码格式。请务必根据需要替换你自己的连接详细信息：DEFINE ROUTE mongo_route WITH TYPE MONGODB ADD MONGODB_CONFIG WITH CONNECTION_STRING "mongodb+srv://<username>:<password>@<cluster-uri>/<database>?tls=true&authSource=admin&replicaSet=<replica-set>" WITH DATABASE "admin"使用来自DigitalOcean的你自己的MongoDB连接详细信息替换，并在你的LoT Notebook中运行该Route。重要提示： 当可用时，请使用VPC连接字符串格式。连接字符串应包括tls=true和authSource=admin参数。有关MongoDB连接故障排除，请参阅我们关于连接MongoDB的教程。为MongoDB数据库存储更新模型修改你的LoT模型以使用数据库路由进行可扩展存储，通过将此添加到模型的末尾：STORE IN "mongo_route" WITH TABLE "MachineProductionData"此外，添加一个带有主题的参数，以便在你的托管数据库中为每个条目提供唯一标识符。DEFINE MODEL MachineData WITH TOPIC "Simulator/Machine/+/Data" ADD "energy" WITH TOPIC "raw_data/+" AS TRIGGER ADD "device_name" WITH REPLACE "+" WITH TOPIC POSITION 2 IN "+" ADD "energy_wh" WITH (energy * 1000) ADD "production_status" WITH (IF energy > 5 THEN "active" ELSE "inactive") ADD "production_count" WITH (IF production_status EQUALS "active" THEN (production_count + 1) ELSE 0) ADD "stoppage" WITH (IF production_status EQUALS "inactive" THEN 1 ELSE 0) ADD "maintenance_alert" WITH (IF energy > 50 THEN TRUE ELSE FALSE) ADD "timestamp" WITH TIMESTAMP "UTC" STORE IN "mongo_route" WITH TABLE "MachineProductionData"部署此更新后的操作后，所有数据在更新时应自动存储在数据库中。OpenSearch集成OpenSearch是一种分布式搜索和分析引擎，专为大规模数据处理和实时分析而设计。在本节中，你将学习如何将你的Coreflux MQTT代理连接到DigitalOcean上的托管OpenSearch数据库，以便你的实时设备数据能够安全可靠地持久化，用于分析、报告或与其他应用程序集成。要启用此集成，你必须在Coreflux的LoT（Language of Things）中定义一个Route，指示处理后的数据应该发送到哪里以及如何发送。下面是路由数据到OpenSearch数据库所需的低代码格式。请务必根据需要替换你自己的连接详细信息：DEFINE ROUTE OpenSearch_log WITH TYPE OPENSEARCH ADD OPENSEARCH_CONFIG WITH BASE_URL "https://my-opensearch-cluster:9200" WITH USERNAME "myuser" WITH PASSWORD "mypassword" WITH USE_SSL TRUE WITH IGNORE_CERT_ERRORS FALSE使用来自DigitalOcean的你自己的OpenSearch连接详细信息替换，并在你的LoT Notebook中运行该Route。重要提示： 当可用时，请使用VPC基础URL（而非公共URL）。基础URL格式通常为https://your-cluster-hostname:9200。对于OpenSearch仪表板访问，请使用数据库集群详细信息中提供的单独的仪表板URL。有关更多详细信息，请参阅我们的OpenSearch快速入门。为OpenSearch数据库存储更新模型修改你的LoT模型以使用数据库路由进行可扩展存储，通过将此添加到模型的末尾：STORE IN "OpenSearch_Log" WITH TABLE "MachineProductionData"此外，添加一个带有主题的参数，以便在你的托管数据库中为每个条目提供唯一标识符。DEFINE MODEL MachineData WITH TOPIC "Simulator/Machine/+/Data" ADD "energy" WITH TOPIC "raw_data/+" AS TRIGGER ADD "device_name" WITH REPLACE "+" WITH TOPIC POSITION 2 IN "+" ADD "energy_wh" WITH (energy * 1000) ADD "production_status" WITH (IF energy > 5 THEN "active" ELSE "inactive") ADD "production_count" WITH (IF production_status EQUALS "active" THEN (production_count + 1) ELSE 0) ADD "stoppage" WITH (IF production_status EQUALS "inactive" THEN 1 ELSE 0) ADD "maintenance_alert" WITH (IF energy > 50 THEN TRUE ELSE FALSE) ADD "timestamp" WITH TIMESTAMP "UTC" STORE IN "OpenSearch_Log" WITH TABLE "MachineProductionData"部署此更新后的操作后，所有数据在更新时应自动存储在数据库中。步骤9 — 验证完整的物联网自动化管道监控实时数据流MQTT Explorer：使用MQTT客户端验证实时数据发布数据库客户端：连接以验证数据的存储（PostgreSQL使用DBeaver，MongoDB使用MongoDB Compass，OpenSearch使用OpenSearch Dashboards）验证PostgreSQL存储使用DBeaver连接到你的PostgreSQL托管数据库以验证可扩展存储：使用来自你的DigitalOcean数据库的连接字符串导航到 coreflux-broker-data 数据库（或你为数据库指定的名称）检查 MachineProductionData 表中存储的记录正如我们之前看到的，所有数据都可在MQTT代理中用于其他用途和集成。验证MongoDB存储使用MongoDB Compass连接到你的MongoDB托管数据库以验证可扩展存储：使用来自你的DigitalOcean数据库的连接字符串导航到 coreflux-broker-data 数据库（或你为数据库指定的名称）检查 MachineProductionData 集合中存储的文档你应该看到具有类似结构的实时数据文档：{ "_id": { "$oid": "68626dc3e8385cbe9a1666c3" }, "energy": 36, "energy_wh": 36000, "production_status": "active", "production_count": 31, "stoppage": 0, "maintenance_alert": false, "timestamp": "2025-06-30 10:58:11", "device_name": "station2" }正如我们之前看到的，所有数据都可在MQTT代理中用于其他用途和集成。验证MySQL存储使用DBeaver连接到你的MySQL托管数据库以验证可扩展存储：使用来自你的DigitalOcean数据库的连接字符串导航到coreflux-broker-data数据库（或你为数据库指定的名称）检查MachineProductionData表中存储的记录与其他集成一样，所有数据也可在MQTT代理中用于其他用途和下游集成。验证OpenSearch存储使用提供的URL和凭据打开OpenSearchDashboards：打开菜单并选择索引管理选项在菜单中选择索引选项，查看你的表名是否出现在列表中返回主页并在菜单中选择发现选项按照提供的步骤创建索引模式返回到发现页面，你应该会看到你的数据正如我们之前看到的，所有数据都可在MQTT代理中用于其他用途和集成。步骤10 - 扩展你的用例和集成测试LoT能力发布示例数据：使用MQTT Explorer将示例数据集发布到你的Coreflux代理。尝试不同的负载结构和不同的模型/操作，查看它们如何处理并存储到你选择的数据库中。数据验证：验证你数据库中的数据与你发布的有效负载是否匹配。使用你的数据库客户端（PostgreSQL使用DBeaver，MongoDB使用MongoDB Compass，OpenSearch使用OpenSearch Dashboards）检查一致性和准确性，确保你的物联网自动化集成按预期工作。比较时间戳、字段转换和数据类型，以验证你的实时数据管道。实时监控：使用另一个MQTT数据源（例如具有MQTT连接功能的简单传感器）设置连续的实时数据馈送。观察Coreflux和你的数据库如何处理传入的物联网数据流，并探索数据检索和查询的响应时间。构建分析和可视化创建仪表板：与Grafana等可视化工具集成，创建显示你的物联网数据的仪表板，从实时MQTT主题和历史数据库查询中提取数据。跟踪指标，如设备正常运行时间、传感器读数、生产计数或来自你自动化系统的维护警报。了解如何使用我们的教程设置DigitalOcean托管数据库与Prometheus和Grafana的监控。对于实时仪表板，直接订阅MQTT主题；对于历史趋势和聚合，查询你的数据库。趋势分析：利用你数据库的能力来分析随时间变化的趋势：PostgreSQL：使用SQL查询进行复杂的关系分析MongoDB：使用聚合框架进行基于文档的分析OpenSearch：使用高级分析和搜索能力进行全文搜索和时间序列分析多数据库集成：探索集成其他托管数据库，如用于非结构化数据的MongoDB，用于关系数据的PostgreSQL，用于结构化查询的MySQL，或用于高级分析和搜索的OpenSearch。使用Coreflux路由将数据同时发送到多个目的地。优化和扩展你的物联网基础设施负载测试：使用LoT Notebook或自动化脚本通过同时发布多条消息来模拟高流量。监控你的Coreflux MQTT代理和数据库集群如何处理负载，并识别你的数据管道中的任何瓶颈。扩展：DigitalOcean提供垂直和水平扩展选项。随着你的物联网数据需求增长，增加droplet资源（CPU、RAM或存储）。扩展你的托管数据库集群以处理更大的数据集，并配置自动扩展警报，以便在接近资源限制时通知你。常见问题解答1. 如何将Coreflux MQTT代理与托管数据库集成？你通过定义指向目标服务（PostgreSQL、MySQL、MongoDB或OpenSearch）的LoTRoute来将Coreflux MQTT代理与托管数据库集成。每个路由使用适当的连接参数（服务器或连接字符串、端口、数据库名称、用户名、密码和SSL/TLS选项），并自动将MQTT消息有效负载持久化到表、集合或索引中。一旦定义好路由，你就使用STORE IN指令将其附加到Model，这样每个处理后的消息都会被写入你选择的数据库。2. 我能否在不编写自定义集成代码的情况下将MQTT数据直接保存到数据库？可以。Coreflux设计为一个低代码集成层，因此你无需编写应用程序代码或外部ETL作业来持久化数据。对于每种数据库类型，你配置一个LoT路由（例如，PostgreSQL_Log、MySQL_Log、mongo_route或OpenSearch_Log），然后使用STORE IN "<route_name>" WITH TABLE "MachineProductionData"扩展你的模型。Coreflux处理连接池、重试和错误处理，因此你可以专注于建模主题和转换，而不是样板数据库代码。3. 我应该为MQTT物联网数据存储选择哪种托管数据库？你的MQTT物联网数据的最佳托管数据库取决于你的数据结构、查询需求和分析目标。使用下面的比较表来帮助你决定：数据库最适合示例用例PostgreSQL强一致性、关系模式、复杂的SQL查询工业传感器网络、事务性事件、需要跨连接数据集的分析MySQL关系数据、结构化查询、广泛的兼容性库存系统、生产指标、传统业务记录MongoDB灵活、不断演进的模式；文档存储具有可变负载的互联设备、具有变化格式的物联网遥测OpenSearch全文搜索、分析、仪表板、日志索引时间序列分析、监控、事件日志、物联网搜索和可视化提示： 你可以通过配置多个Coreflux路由同时使用多个托管数据库。这使得可以从同一个MQTT流中，将结构化的物联网数据存储在PostgreSQL或MySQL中，在OpenSearch中聚合日志和指标，并在MongoDB中收集非结构化或无模式数据。4. 这种架构如何处理实时和历史分析？Coreflux将所有处理后的值保留在MQTT主题上，供实时消费、仪表板或额外管道使用，而Routes则将相同的建模数据持久化到你的数据库中，用于历史查询。在实践中，你可以订阅主题以进行即时反应（警报、控制回路），并查询PostgreSQL/MySQL/MongoDB/OpenSearch以进行聚合、趋势和长期分析。这种双路径设计反映了MQTT和物联网数据集成教程中的常见模式，其中代理提供实时消息传递，而数据库提供持久存储和分析。5. Coreflux和托管数据库之间的连接有多安全？当部署在DigitalOcean上时，你可以使用VPC网络来保持Coreflux MQTT代理和数据库之间的所有通信私密。VPC将你的资源与公共互联网访问隔离开来，并且DigitalOcean托管数据库支持连接的TLS加密。此外，你可以为你的Coreflux应用程序创建具有有限权限的专用数据库用户，遵循最小权限原则。6. 这个设置是否适用于生产环境物联网部署？是的。这种架构反映了生产环境中MQTT和数据库集成所使用的模式，其中代理前端处理设备流量，而托管数据库层提供持久性和分析。DigitalOcean托管数据库提供自动备份、高可用性和监控，而Coreflux MQTT代理可以水平扩展以处理高消息吞吐量。对于生产环境，你还应该配置防火墙规则、使用强凭据、为MQTT和数据库连接启用TLS，并根据预期的消息量来调整你的droplet和集群大小。7. 我能否在没有公共互联网访问的情况下，或在混合环境中运行MQTT代理？可以。MQTT代理通常部署在私有网络或边缘环境中，公共资源一致指出，只要客户端可以访问代理，MQTT就可以在没有公共互联网的情况下工作。使用DigitalOcean，你可以将Coreflux和你的数据库保持在VPC内部，并且只暴露绝对必要的内容（例如，VPN、堡垒主机或有限的防火墙规则）。如果你需要混合或多站点架构，你还可以将选定的主题与其他代理或云区域同步。8. 在物联网数据中使用MQTT和数据库是否存在任何限制或权衡？MQTT针对轻量级、事件驱动的消息传递进行了优化；数据库则针对存储和查询进行了优化。存储每一条原始消息可能会变得昂贵或嘈杂，因此最佳实践建议仔细建模数据（例如，聚合指标、过滤主题或降采样）。极低功耗设备或超受限网络可能难以维持持久连接或处理TLS开销，在这种情况下，你可能需要调整QoS级别、批处理和保留策略。只要你在设计中考虑到这些权衡，MQTT加上托管数据库对于大多数物联网场景都能很好地工作。9. 我如何为我的物联网项目在PostgreSQL、MySQL、MongoDB和OpenSearch之间做出选择？你应该根据物联网数据结构、可扩展性以及你希望如何查询设备数据来选择托管数据库。下表总结了每个选项的优势：数据库当...时最佳典型用例关键优势PostgreSQL你需要复杂的关系查询、强一致性和事务完整性（ACID支持）。工业传感器网络、将设备数据与生产相关联、需要对连接的数据集进行分析关系模式、高级SQL、一致性MySQL你的工作负载是结构化的，具有广泛的工具和兼容性需求。库存跟踪、传统业务系统、生产指标更简单的关系需求、广泛支持MongoDB你的设备负载和模式不断演变，或者你希望使用灵活的、基于文档的存储进行快速原型设计。具有可变格式的物联网遥测、快速开发、半结构化数据灵活的模式、易于扩展、快速原型设计OpenSearch你需要分析、搜索或对大容量的物联网数据（日志、时间序列、事件）进行仪表板展示。搜索传感器数据、日志分析、可视化、基于关键字/时间的查询搜索、全文、分析、快速聚合结论将Coreflux MQTT代理与DigitalOcean的托管数据库服务（PostgreSQL、MongoDB、MySQL或OpenSearch）集成，为你提供了实时物联网数据处理和存储的完整设置。按照本教程，你已经使用低代码开发实践构建了一个收集、处理和存储物联网数据的自动化管道。借助Coreflux的架构和你选择的数据库的存储特性，你可以处理大量的实时数据并查询它以获取洞察。无论你是监控工业系统、跟踪环境传感器还是管理智慧城市基础设施，这种设置都让你能够基于实时MQTT主题和历史数据库查询做出数据驱动的决策。了解更多关于DigitalOcean托管数据库的信息，以及DigitalOcean 针对 IoT行业的产品服务支持，可咨询 DigitalOcean 中国区独家战略合作伙伴卓普云AI Droplet（aidroplet.com）。你可以尝试提供的用例或使用Coreflux和DigitalOcean实现你自己的用例。你还可以在DigitalOcean Droplet市场或通过Coreflux网站获取免费的Coreflux MQTT代理。
前面的章节（社区专栏《SQL调优》）我们已经写了很多篇幅关于 MySQL 执行计划的解读，今天我们来继续延伸介绍执行计划的链路跟踪功能，也就是 MySQL 的 Optimizer Trace。在这之前，先来回顾下 EXPLAIN 的结果：mysql:ytt>explain select * from t1 a left join y1 b on a.id = b.id where a.r1<100 order by a.r2 desc; +----+-------------+-------+------------+--------+---------------+---------+---------+----------+--------+----------+-----------------------------+ | id | select_type | table | partitions | type | possible_keys | key | key_len | ref | rows | filtered | Extra | +----+-------------+-------+------------+--------+---------------+---------+---------+----------+--------+----------+-----------------------------+ | 1 | SIMPLE | a | NULL | ALL | idx_r1 | NULL | NULL | NULL | 998222 | 50.00 | Using where; Using filesort | | 1 | SIMPLE | b | NULL | eq_ref | PRIMARY | PRIMARY | 4 | ytt.a.id | 1 | 100.00 | NULL | +----+-------------+-------+------------+--------+---------------+---------+---------+----------+--------+----------+-----------------------------+ 2 rows in set, 1 warning (0.00 sec)EXPLAIN 展示出来的核心数据有：表关联顺序优化器筛选过的索引实际使用的索引每张表依据统计信息的扫描行数Extra 额外数据提示两种执行计划（explain format=tree / explain format=json）展示出来的额外成本数据如果想快速对于 SQL 进行优化，基于以上的结果完全可以满足。但是想深入了解 MySQL 优化器为什么选择这样的执行计划，基于以上的结果就无法满足。举例说明：我想知道对于表 a 来讲，为什么有索引 idx_r1，但是实际却没有使用，而走的全表扫？两张表关联，为什么选择的顺序是表 a 驱动表 b，而不是表 b 驱动表 a？为什么字段 r2 有索引，但是依然要走排序？带着这些疑问，我们来介绍 MySQL 的 Optimizer Trace 功能。1. 什么是 Optimizer Trace？简单来讲，Optimizer Trace 是一个 SQL 执行计划的链路跟踪器，跟踪 SQL 的解析、优化、执行等过程，并且把结果记录到 MySQL 元数据表（information_schema.optimizer_trace），之后可以对这张表分析得到很多个执行计划的“为什么？”！2. 如何使用 Optimizer Trace？要使用 Optimizer Trace 功能，首先得打开控制开关。谨记：这个功能非常耗费资源，默认关闭的，可以通过调整以下变量开启：mysql:ytt>show variables like 'optimizer_trace%'; +------------------------------+----------------------------------------------------------------------------+ | Variable_name | Value | +------------------------------+----------------------------------------------------------------------------+ | optimizer_trace | enabled=off,one_line=off | | optimizer_trace_features | greedy_search=on,range_optimizer=on,dynamic_range=on,repeated_subselect=on | | optimizer_trace_limit | 1 | | optimizer_trace_max_mem_size | 1048576 | | optimizer_trace_offset | -1 | +------------------------------+----------------------------------------------------------------------------+ 5 rows in set (0.00 sec)以上几个参数详细解释下：optimizer_trace：enabled=on/off 启用/禁用 Optmizer Trace 功能；one_line=on/off 启用/禁用 json 格式化存储，一般不需改动。optimizer_trace_limit/optimizer_trace_offset：这两个参数和 LIMIT 子句一样，用来最终展示 Trace 的 SQL 条数。展示的条数越多，对内存消耗越大，默认展示最近的一条记录。比如设置 optimizer_trace_limit 为 10，optimizer_trace_offset 为 -10，就可以最多展示 10 条 Trace 记录。optimizer_trace_max_mem_size：用来存储 Trace 结果的最大内存。optimizer_trace_features：用来启动/禁用相关 Trace 特性开关。end_markers_in_json：启用/禁用 注释功能。开启这个，Trace 结果可读性更强。Optimizer Trace 可以跟踪的语句有：SELECT、TABLE、VALUES、WITH、INSERT、REPLACE、UPDATE、DELETEEXPLAINSET（排除设置 Optimizer Trace 相关参数）DO存储函数内部、触发器内部等的 DECLARE、CASE、IF、RETURN 语句CALL在数据库里，语句调优一般说的是 SELECT 语句，所以大部分场景跟踪的也只有 SELECT 语句。元数据表字段解析mysql:ytt>desc information_schema.optimizer_trace; +-----------------------------------+----------------+------+-----+---------+-------+ | Field | Type | Null | Key | Default | Extra | +-----------------------------------+----------------+------+-----+---------+-------+ | QUERY | varchar(65535) | NO | | | | | TRACE | varchar(65535) | NO | | | | | MISSING_BYTES_BEYOND_MAX_MEM_SIZE | int | NO | | | | | INSUFFICIENT_PRIVILEGES | tinyint(1) | NO | | | | +-----------------------------------+----------------+------+-----+---------+-------+ 4 rows in set (0.00 sec)QUERY：TRACE 的 SQL 语句原文TRACE：SQL 语句的 TRACE 结果，JSON 格式存储（由变量 end_markers_in_json 来控制）MISSING_BYTES_BEYOND_MAX_MEM_SIZE：TRACE 结果超过变量 optimizer_trace_max_mem_size 设置的值后，截断的大小（BYTE）INSUFFICIENT_PRIVILEGES：对存储过程、存储函数等包含有 SQL SECURITY DEFINER 的用户是否有对应的权限，有权限为 0，无权限为 1，并且 TRACE 字段为空。Optimizer Trace 开启步骤mysql:ytt>set optimizer_trace='enabled=on'; Query OK, 0 rows affected (0.00 sec) mysql:ytt>set optimizer_trace_limit=10; Query OK, 0 rows affected (0.00 sec) mysql:ytt>set optimizer_trace_offset=-10; Query OK, 0 rows affected (0.00 sec) mysql:ytt>set end_markers_in_json=on; Query OK, 0 rows affected (0.00 sec)这里要注意的是，修改任何一个 Optimizer Trace 相关参数，元数据表 information_schema 表都会被清空。mysql:ytt>select count(*) from information_schema.optimizer_trace; +----------+ | count(*) | +----------+ | 10 | +----------+ 1 row in set (0.00 sec) mysql:ytt>set optimizer_trace_offset=-2; Query OK, 0 rows affected (0.00 sec) mysql:ytt>select count(*) from information_schema.optimizer_trace; +----------+ | count(*) | +----------+ | 0 | +----------+ 1 row in set (0.00 sec)3. Optimizer Trace 的结果我们用一个最简单的例子来看看 Optimizer Trace 的大致结构：do 语句非常简单，只用来验证是否语法正确，不出结果。mysql:ytt>do 1+1; Query OK, 0 rows affected (0.00 sec)下面是 Optimizer Trace 结果：mysql:ytt>select query,trace from information_schema.optimizer_trace\G *************************** 1. row *************************** query: do 1+1 trace: { "steps": [ { "join_preparation": { "select#": 1, "steps": [ { "expanded_query": "/* select#1 */ select (1 + 1) AS `1+1`" } ] } }, { "join_optimization": { "select#": 1, "steps": [ ] } }, { "join_execution": { "select#": 1, "steps": [ ] } } ] } 1 row in set (0.00 sec)可以看到，Optimizer Trace 结果是一个 JSON 串，key 为 steps，value 是一个数组，数组有三个 key，分别为：join_preparation 准备阶段：这里会做一些 SQL 改写，关键字识别等等，可以看到 expanded_query 对应的值即为 SQL 语句被改写后的内部 SQL。join_optimization 优化阶段：具体 SQL 优化，包括一些可能的逻辑优化，一些根据表统计信息预估的物理优化等等。join_execution 最终执行阶段：最终 SQL 采用的执行计划等等。本篇是Optimizer Trace的开端，由于内容太多，我特地拆分为几篇来写，欢迎继续订阅。
爱奇艺基于OceanBase实现百亿级卡券业务的“单库双擎”架构升级
摘要：爱奇艺卡券业务原采用 “MySQL 分库分表 + ES 异步同步” 架构，面临 TP/AP 分离导致的架构复杂、AP 查询分钟级延迟、数据一致性隐患等问题。如今借助 OceanBase 的 HTAP 能力，将 AP、TP 业务融合到一个数据库，在架构简化、成本控制与效率提升方面均取得了突破。爱奇艺是国内知名的在线视频平台，每年都会推出上百部优质长视频内容，其中不乏如 2023 年现象级爆款《狂飙》这样的佳作。近两年，随着短视频、微剧的兴起，平台年处理视频内容数量级从上百部直接跃升至上万部。每一部内容从立项、拍摄、生产、制作到上线播出，均需经过复杂流程，并依赖上百个业务模块协同支持。卡券业务是出于整个业务生态里面的中台位置，对上提供中台能力，如为创新型业务会员、云影院提供商业化变现和用户转化营销工具。过去，卡券业务系统将 AP（分析处理）与 TP（事务处理）业务分开处理，架构复杂，需要较高的维护成本；如今，借助 OceanBase 的 HTAP 能力，将 AP、TP 业务融合到一个数据库，在架构简化、成本控制与效率提升方面均取得了突破。解构卡券业务的数据架构困境卡券是爱奇艺核心的营销与促销工具，贯穿爱奇艺的会员购买、云影院观影等商业化变现全链路。其底层数据库的性能直接关乎用户体验与业务敏捷性。例如，促销发券、会员领券等场景均需要业务系统提供高并发事务处理（TP）能力。而在运营侧，则需要实时统计和分析发券数量、会员领券等指标，以便评估活动效果、优化营销策略，这依赖于高效的数据分析（AP）能力。过去，卡券业务系统采用的是“MySQL 分库分表+ES 异步同步”的复杂架构：分库分表的 MySQL 来承载 TP 业务，以应对高并发海量请求；Elasticsearch（ES）来完成统计分析等 AP 类业务。“这个卡券业务的架构基本上能为业务需求提供足够的支撑。不过，我们并不满意，一直在寻找新的解决方案。”爱奇艺高级总监张冲表示。其最重要的原因就在于系统架构过于复杂，虽能满足业务需求，但每一个节点都需要研发投入大量精力维护，顶峰的时候研发资源占比近 80%，严重挤占了业务创新资源。具体而言，在 TP 业务方面，通过分库分表的 MySQL 集群支撑高并发交易，但实际日常资源利用率仅约 10%，资源冗余明显。在 AP 业务方面，ES 进行运营统计分析时的数据源自订阅多个 MySQL 实例的 binlog，经消息队列 RMQ 异步同步至 ES，链路冗长，存在分钟级延迟。并且 ES 的清理归档代价较高，Reindex 开销也比较大。此外，数据的一致性与准确性面临挑战，在异步同步过程中，甚至出现过 UV（访客数量）超过页面点击的异常，统计准确率难以保障。张冲坦言，对现有架构进行升级更深层的原因，源于对技术进步的持续追求。“我们希望技术上再往前走一走，要和互联网行业最先进的技术保持对齐。”从“分库分表+ES”到“单库双擎”，爱奇艺 HTAP 架构升级实践随着新技术趋势的出现，爱奇艺也开始寻找能够简化架构、支撑业务更好发展的新发展。数据库的升级是这次架构升级的关键。对于新一代数据库，爱奇艺提出了明确的要求：第一，必须是一款兼顾 TP、AP，具备 HTAP 能力的数据库产品，无需管 MQ，无需处理异构的数据，尽量减少对数据平台的依赖，以简化数据底座；第二，总体成本可控，和现有架构相比成本不能上升，符合公司降本增效的目标；第三，云中立，在遇到故障的时候可以实现云逃逸，且在不同的云上均可提供一致性的服务。根据上述三个基本要求，经过对多款主流数据库产品的调研与测试，OB Cloud 一体化云数据库凭借其卓越的性能与高度契合的需求满足度，赢得了爱奇艺的青睐，并且在高并发、高可用、安全、数据治理、低成本等方面的技术积累，都被浓缩到 OB Cloud 一体化云数据库的产品中。张冲表示：“OceanBase 不仅提供了真正的 HTAP 融合能力，其原生分布式架构还与我们的云原生战略高度契合。同时，OB Cloud 在百度云上开服也是一个重要契机，因为爱奇艺的系统平台就部署在百度云上。”完成数据库选型之后，爱奇艺迅速开始了数据和架构升级的准备工作。升级工作分为两个阶段：AP 升级：将 ES 集群中的百亿文档升级至 OB Cloud 集群。通过双写、迁移历史数据、切读、停双写等步骤，不仅完成数据升级，还从业务层面进行了逻辑去冗余和简化。最终，资源成本下降 60%，运营查询类 SQL 基本在 1 秒内返回。TP 升级：将 16 个物理机数据库从原生 MySQL 分库分表形态升级至 OB Cloud 集群。借助 OceanBase 的 OMS 同步工具，顺利完成海量数据同步与校验工作。最终，存储成本下降 80%，且系统具备弹性伸缩能力，无需为大促提前预备资源。张冲表示，为了尽量减少对业务的干扰，保持业务稳定性，升级过程尽可能少地修改代码，他们采取了一些措施：汇聚到 OceanBase 的分区表、分区键与原来的分片逻辑一致，使得业务系统零改造切换；保持 AP 业务不变，仅修改数据源订阅，通过全兼容的 binlog 直接订阅到同租户的 AP 表。此时还是多份存储，但依靠高压缩比，整体存储成本没有上升；将 TP 业务的表异步修改为行列混存，不影响业务稳定性，同时运营统计只需简单修改库和表名即可快速上线。通过多副本读写分离，最终实现了单库双擎、支撑实时在线业务与数据分析的简洁架构。化繁为简，打造卡券业务的现代化数据库底座经过升级改造后，卡券业务系统架构变得非常简洁，只有基本的业务服务和数据中台的数据交互，不再需要维护额外的数据流服务，也无需担心存储不足等问题，归档清理的周期也相应延长，研发人员得以更加聚焦于业务需求的开发。张冲表示，卡券业务系统架构升级带来了如下好处：首先，链路极大简化。 去除了 MySQL 到 ES 的异步同步链路，消除了 ES 集群的运维与成本；张冲特别感慨此次架构的升级带来的简化，他表示“简单到只有计算、只有存储，简单到有点像互联网刚开始发展的那个阶段。”其次，分析效率提升。 常规 AP 查询可直接在 OB Cloud 中完成，时间也从原来的准实时变成了实时，所有统计 SQL 的响应时间（RT）均小于 1 秒，性能大幅提升。而 BI 类需求以前需要数据中台部门支持，属于跨部门协作，最快也需数天；现在本部门内部就能完成，时间缩短为 2-3 天；第三，存储成本显著节省。 借助 OceanBase 的高压缩能力，相比 MySQL 的存储成本下降了 80%，并且它可以弹性伸缩，不再需要提前为大促预备过量资源。“相对于之前的架构，现在的架构非常简洁。这要得益于 OceanBase 把高并发、高可用、安全、数据治理、低成本等各种技术积累都浓缩到了这个数据库产品中。”张冲这样评价。小结张冲表示，未来爱奇艺计划将 OceanBase 的实践推广至更多在线交易型业务，如订单支付、会员中心等，并逐步探索其在 KV 存储、向量检索等场景的应用。面对 AI 浪潮，张冲也提出了对未来数据库的期待：“知识图谱、AI 工作流等复杂场景，需要更智能的底层数据支撑。希望 OceanBase 能在这些方向持续演进，成为企业智能化转型的数据基石。借助 OceanBase 技术的不断完善和应用场景的拓展，爱奇艺将在科技创新的道路上走得更远。”欢迎访问 OceanBase 官网获取更多信息：https://www.oceanbase.com/
实现设备监控与风速告警的实时化升级，山东港口科技借助时序数据库 TDengine 构建智慧港口“数据基石”
小T导读：在智慧港口的建设过程中，面对海量物联网设备产生的时序数据（如设备状态、能耗、作业效率等）的高效接入与实时分析需求，山东港口科技选择采用 TDengine TSDB 时序数据库作为核心数据底座，以应对传统关系型数据库在处理高并发、大规模时序数据时的性能瓶颈，实现设备状态的实时监控、数据压缩存储与智能分析，为智慧港口的数字化转型与智能化运营提供强有力的数据支撑。本次将就此实践进行具体分享。合作背景在“智慧港口”的宏伟蓝图下，山东港口科技集团面临着海量物联网设备数据接入、处理与分析的严峻挑战。港口作业涉及大量的桥吊、门机、集卡、传感器等终端设备，这些设备 7x24 小时不间断产生巨量的时序数据（如位置、状态、能耗、效率指标等）。传统的通用关系型数据库在处理这类高并发、海量的时序数据时，显得力不从心。为了夯实智慧港口的数据根基，经过严谨的选型，我们最终选择了 TDengine TSDB 作为核心时序数据平台，以支撑关键业务系统的数字化转型。选择 TDengine TSDB 的原因在引入 TDengine TSDB 之前，我们的业务系统主要面临以下痛点：数据膨胀与存储成本高：​ 港口设备每秒产生数以万计的数据点，若采用传统数据库存储，数据表会急剧膨胀，不仅占用大量存储空间，且备份和维护成本高昂。查询分析效率瓶颈：​ 对于实时监控、效率分析和历史数据回溯等场景，传统数据库的查询响应速度慢，无法满足业务对“实时洞察”的要求，特别是在聚合计算大量设备的历史数据时，耗时长达分钟甚至小时级。系统架构复杂：​ 为了应对不同的数据处理需求（如实时、短期、长期），往往需要组合使用多种数据库和技术栈（如 Redis、MySQL、Hadoop 等），这增加了系统架构的复杂性、开发和运维难度。TDengine TSDB 作为专为时序数据设计的数据库，其超高性能、内置缓存和流式计算功能、极简的架构以及强大的数据压缩能力​，恰好精准地解决了上述痛点，成为我们的理想选择。使用 TDengine TSDB 后的收益与业务提升部署 TDengine TSDB 后，我们在多个方面获得了显著收益：极致的性能提升：​ 对港口设备运行状态的查询响应速度从原来的“分钟级”提升到“毫秒级”，实现了真正的实时监控与告警。显著的降本增效：TDengine TSDB 高效的数据压缩技术，使得存储空间节省超过 80%，大幅降低了硬件与运维成本，简化的架构也减少了运维团队的工作负担。增强的数据驱动能力：​借助 TDengine TSDB 强大的时序数据计算能力，业务团队能够轻松进行设备效率分析、预测性维护和运营优化，为决策提供坚实的数据支持，进一步强化了“智慧港口解决方案”的核心优势。加速创新应用落地：借助 TDengine TSDB 这一稳定的高性能数据底座，我们能够快速开发和部署新的数据密集型应用，如全自动码头的智能调度系统、物流供应链的可视化平台等。核心业务场景与 TDengine TSDB 应用实例场景一：港口岸桥设备实时状态监控与效率分析业务描述：​ 实时监控码头所有岸桥（Quay Crane）的运行状态（如起升、下降、大车行走、小车行走）、能耗以及作业效率（如单箱能耗、作业周期），确保设备安全高效运行，并即时发现异常。TDengine TSDB 查询 SQL 示例：-- 1. 查询指定岸桥（Crane_ID = 'QC08') 在过去10分钟内的平均功率和总能耗 SELECT AVG(power_kw), SUM(power_kw * ts_interval / 3600) AS total_energy_kwh FROM crane_power_metrics WHERE crane_id = 'QC08' AND ts >= NOW - 10m INTERVAL(1m); -- 2. 统计过去1小时内，所有岸桥的作业箱量（基于每次吊装动作计数） SELECT crane_id, COUNT(*) AS operation_count FROM crane_operation_events WHERE ts >= NOW - 1h AND operation_type = 'lift_complete' GROUP BY crane_id;​通过 TDengine TSDB 毫秒级查询与高效聚合能力，我们实现了对数百台岸桥设备运行状态的实时监控（1 分钟粒度）与异常秒级捕捉，查询效率从分钟级提升至毫秒级，存储成本降低超 80%，极大提升了设备管理实时性与安全性。场景二：智能集卡（AGV/IGV）调度与路径优化业务描述：​ 追踪自动化码头内数百台智能导引车（AGV）的实时位置、速度、电池电量和状态，基于这些时序数据进行最优路径规划和调度，避免拥堵，提升整体物流周转效率。TDengine TSDB 查询 SQL 示例：​-- 1. 查询所有电量低于20%的AGV的当前位置和最新电量 SELECT last(latitude), last(longitude), last(battery_level) FROM agv_status_metrics WHERE battery_level < 20 GROUP BY agv_id; -- 2. 计算指定区域（如A01区）过去5分钟内的平均车辆速度，用于判断拥堵情况 SELECT AVG(speed_kmh) AS avg_speed FROM agv_location_metrics WHERE ts >= NOW - 5m AND zone_id = 'A01';借助 TDengine TSDB 的 last() 实时状态查询与窗口聚合能力，我们实现了对数百台 AGV 的实时位置、电量及速度监控，低电量车辆识别与区域拥堵判断均达到秒级响应，调度效率提升约 50%\~70%，整体物流周转更高效、更智能。场景三：港口风速风向监测与预警业务描述：​分布在港区各处的气象站持续采集风速、风向数据。系统需要实时判断是否超过安全作业阈值，并及时向相关设备和人员发出预警，保障恶劣天气下的作业安全。TDengine TSDB 流计算 SQL 示例：​-- 创建流式计算，持续监控风速，一旦发现某个站点每分钟一次的平均风速超过阈值（18m/s），则触发告警 CREATE STREAM wind_alert_stream INTO wind_alert_events AS SELECT _wstart AS ts, station_id, AVG(wind_speed) AS avg_wind_speed FROM weather_station_metrics PARTITION BY station_id INTERVAL(1m) SLIDING(1m); -- 查询历史告警记录 SELECT * FROM wind_alert_events WHERE ts >= TODAY ORDER BY ts DESC;解析如下：CREATE STREAM wind\_alert\_stream 定义了一个名为 wind_alert_stream的流，用于持续处理实时数据。INTO wind\_alert\_events 将流计算的结果写入到 TDengine TSDB 中的 wind_alert_events表中，该表为一个超级表，按照分组会自动生成子表，用于存储每个分组的告警事件。SELECT \_wstart AS ts, station\_id, AVG(wind\_speed) AS avg\_wind\_speed 选择数据流中的时间戳（\_wstart）、站点 ID（station\_id）以及风速的平均值（AVG(wind\_speed)）。_wstart是该时间窗口的起始时间，作为告警触发的时间点。FROM weather\_station\_metrics 数据源是 weather_station_metrics表，该表应包含字段如：ts（时间戳）、station_id（站点 ID）、wind_speed（风速-单位：m/s）等。PARTITION BY station\_id 按站点分组，每个站点独立计算，避免不同站点之间的数据干扰。INTERVAL(1m) SLIDING(1m) 定义了 1 分钟的时间窗口，每 1 分钟滑动一次，即每分钟统计一次过去 1 分钟内的数据。借助 TDengine TSDB 灵活的流计算能力（1 分钟滑动窗口），我们实现了港口风速的实时监测与自动告警（响应时间＜1 分钟）。原本需要多个大数据组件才能完成的处理流程，如今只需一条语句即可完成，告警的准确性与时效性显著提升，安全运维效率也随之大幅提高。结语通过引入 TDengine TSDB，我们成功构建了一个高性能、高可用的时序数据管理平台，有效解决了智慧港口建设中海量物联网数据处理的核心难题。这一合作不仅提升了现有业务的运营效率和智能化水平，也为未来探索更多基于数据的创新应用（如数字孪生港口）奠定了坚实的基础，有力地支撑了山东港口科技集团有限公司打造“行业领先的高新技术上市企业”的战略目标。关于山东港口科技山东港口科技集团有限公司是山东省港口集团为全力推进智慧港口建设而设立的高科技子公司。公司立足信息化顶层设计、核心应用系统研发和大数据应用，致力于打造物流供应链服务平台、智慧港口解决方案和自动化应用系统三大核心优势。作为一家以创新为驱动的高新技术企业，科技集团正积极利用数字技术，为全球港口行业的智能化升级注入科技力量。作者：张艳明
主流关系型数据库系统缺陷实证研究——OceanBase 校企联合研究
摘要：OceanBase联合中国人民大学数据库团队的数据库缺陷实证研究，被软件工程顶刊IEEE TSE录用。该研究首次构建了面向开源关系型数据库的细粒度缺陷分类体系，共获得12 项发现，为RDBMS系统的开发维护和测试提供了重要启示。研究发现，涉及SQL数据类型及数据库触发器、存储过程、参数设置等复杂功能的缺陷现有测试工作无法有效触发，这一发现为提升RDBMS缺陷检测能力提供了显著改进空间。日前，由 OceanBase 联合中国人民大学数据库系统研究团队（刘爽副教授）对主流关系型数据库系统缺陷开展的实证研究《A Comprehensive Study of Bugs in Relational DBMS》被软件工程领域顶级期刊 IEEE Transactions on Software Engineering(TSE) 正式录用。IEEE TSE 是 IEEE 旗下软件工程方向的权威期刊。在数据库缺陷实证研究方面，本论文首次系统分析了 MySQL 等三个开源数据库中 777 个真实缺陷，揭示了 RDBMS 的缺陷在根因、表症等方面的特点，以及现有测试工具在深层语义缺陷检测上的局限性。以下为论文介绍。简 介本研究通过“系统性实证分析”揭示主流关系型数据库在真实场景中的缺陷规律。研究覆盖 MySQL、SQLite 和 openGauss 三大系统中 777 个高质量修复缺陷，深入剖析其根本原因、症状表现、分布特征及其关联性。其核心贡献在于：首次构建了面向开源关系型数据库的细粒度缺陷分类体系，研究共获得 12 项发现，为 RDBMS 系统的开发维护和测试提供了重要启示。研究发现，涉及 SQL 数据类型及数据库触发器、存储过程、参数设置等复杂功能的缺陷现有测试工作无法有效触发，这一发现为提升 RDBMS 缺陷检测能力提供了显著改进空间。方法与分类体系表1：采集 bug 的统计信息本研究通过一套严谨的实证方法对关系型数据库中的真实缺陷进行系统性分类与归因。围绕三个核心维度展开：根因、症状和修复模块。研究团队从 MySQL、SQLite 和 openGauss 的官方仓库中收集了 2018 至 2023 年间报告的 2495 个缺陷，经过严格筛选后构建了一个高质量的 777 个缺陷数据集。在此基础上，作者提出了一套四维分析框架：根因维度识别出 12 类根本问题（如错误逻辑、API 误用、类型处理缺陷等）；症状维度归纳了包括错误结果、崩溃、死锁、性能退化等行为；模块维度定位缺陷修复位置（如解析器、优化器、执行引擎、存储层等）；关联性进一步探索三者之间的关联规律，例如“类型相关根因多导致错误结果，且集中于表达式求值模块”。为确保标注一致性，两名研究人员独立完成全部标签分配，并通过 Cohen’s Kappa 系数评估达成共识。该方法不仅保证了分析的客观性，也为后续数据库测试工具的设计提供了可操作的指导依据。结果与分析研究揭示了多项关键发现。首先，在根因分布上，“不正确的代码逻辑”占比最高达 32.3%，“类型处理缺陷”和“API 误用”分别以 9.0% 和 8.4% 的比例成为第二、第三大类根因。其次，在症状表现方面，“结果不一致”是最普遍的症状，占全部缺陷的 42.99%，且往往无崩溃、无报错，具有极强的隐蔽性。图1：按根本原因划分的缺陷分布进一步的跨系统对比显示：MySQL 与 SQLite 在缺陷模式上高度相似，而 openGauss 因架构差异与活跃开发状态，表现出显著不同的缺陷谱系。这些结果不仅刻画了数据库内核的脆弱面，也为未来高可靠数据库的设计与质量保障工作指明了方向。图2：症状与根因的关系概念验证工具 SQLT研究中观察到类型相关缺陷在数据库 bug 中占比显著，团队开发了一个概念验证工具 SQLT，用于针对性挖掘此类问题。SQLT 强化了对跨数据类型表达式、隐式类型转换以及非标准类型（如 BIT、JSON）组合的查询生成能力。该工具通过比对语义等价查询的执行结果，能够有效识别那些不触发崩溃但返回错误结果的静默逻辑缺陷。在实验中，SQLT 不仅成功复现了多个已知类型 bug，还新发现 8 个此前未被报告的问题，其 5 个已被 MySQL、SQLite 和 openGauss 官方确认并修复。表2：SQLT检测到的缺陷欢迎访问 OceanBase 官网获取更多信息：https://www.oceanbase.com/
【文档悬赏令】第2号：数据库巡检手册征集，上传文档赢好礼！
墨天轮社区举办的 【文档悬赏令】系列活动，每次聚焦一个主题，征集真实、可操作的第一手文档，希望能扩充社区文档资源、为更多人提供有用的参考资料，让技术学习少走弯路！第2号悬赏令活动主题：数据库巡检方案实践数据库巡检是保障业务稳定运行的核心运维环节，更是DBA日常工作的重中之重。当前数据库类型、场景十分丰富，多数从业者需耗费大量时间整理巡检清单、调试巡检流程。墨天轮社区第二期【文档悬赏令】活动特围绕这一主题发起有奖征集活动，诚邀您上传实用、可落地的数据库巡检文档，和社区众位DBA们互帮互助，让巡检工作更高效、更标准！一、活动时间2026 年 1 月 26 日 - 3 月 28 日二、文档要求1、主题范围本次征集聚焦 “数据库巡检” 核心主题，覆盖国内外主流数据库，基础巡检、专项巡检、自动化巡检等多类实用方案等均可，具体范围如下：巡检主题细分（包含但不限于）：基础巡检：日常运维巡检表/巡检清单/巡检手册专项巡检：性能监控与优化巡检/灾备状态巡检/单机or集群巡检方式：手动巡检标准化文档/自动化巡检/脚本命令/巡检工具数据库不限：商业数据库：Oracle、SQL Server等国产数据库：达梦DM8、人大金仓KingbaseES、OceanBase等国产库开源数据库：MySQL、PostgreSQL、Redis等开源库2、合格要求与格式文档内容需明确巡检目的、适用场景、巡检流程等，包含巡检相关代码（部分敏感信息可隐去）。页数要求：≥5 页必加标签：上传时需在 “标签” 栏填写 “数据库巡检” + 数据库种类（如 Oracle、OceanBase） 两个标签支持格式：优先推荐 .doc、.pdf、.ppt、.md、.txt 格式（可支持前 5 页预览，下载率更高）；也可上传 .zip （需附内容说明）；以下文档将被判为不合格：1）主题无关：非“数据库巡检”相关主题2）内容搬运：直接上传电子书或产品官方文档、他人演讲PPT，或直接复制抄袭、全文搬运网站其他文章/文档内容（已发布的文章内容不可以同时上传成文档参与活动，但如果是您曾发在其他网站的内容则可以上传参与）3）流水账或凑字数：文档页数需≥5页，但不可通过凑字数、使用大量无关图片占篇幅等4）作者刷量：不可刷数据-我们鼓励作者自发宣传自己的文档，但不可使用人工刷量、注册小号刷量等方式提高下载量；不可将文档拆分多份上传，导致单份内容无实际意义、缺乏完整性5）重复文档：重新上传以前发布的文档系统会自动识别判定为重复仅自己可见，也不可删除旧文档后重新上传三、上传步骤登录上传：登录墨天轮账号后，点击链接直达上传页：https://www.modb.pro/docUpload，或在首页下拉框点击 “传文档”；填写信息：标题：建议明确标注 “数据库种类 + 巡检场景”（如《MySQL数据库巡检手册》《Oracle数据库常规巡检项目和命令》），帮助他人快速判断主题标签：上传时需在 “标签” 栏填写 “数据库巡检” + 数据库种类（如 Oracle、OceanBase） 两个标签墨值设置：支持免墨值 / 5 墨值 / 10 墨值 / 25 墨值 / 50 墨值 / 100 墨值，设置墨值后，他人下载时支付的墨值将实时计入您的账户（ps：一般来说墨值越低下载门槛越低，被下载可能更高）提交：确认信息无误后点击 “提交文档”即可。于 “控制台-内容管理-我的文档” 处可查看您所有文档明细，若您遇到“仅自己可见”“转换失败”等状态可询问墨天轮小助手（VX：modb666）询问具体原因。四、奖励设置本次活动奖励分为 “合格奖”“有效助人奖”“巡检先锋奖” 三类1、合格奖每位用户每日首次上传≥5 页的合格文档，可自动触发 “每日任务” 获 5 墨值。此外，本次活动奖励额外可叠加，根据用户合格文档数量发放不同等级的墨值奖励，具体如下：合格数量墨值0-3个10墨值/份3个以上15墨值/份2、有效助人奖根据单个文档下载数发放不同等级的墨值奖励，每个用户最多可有5个文档获得本项奖励：被下载数墨值30＜被下载数 ≤ 5050墨值50＜被下载数 ≤ 100100墨值100＜下载数 ≤ 150200墨值150＜下载数 ≤ 200300墨值下载数＞200500墨值3、巡检先锋奖将综合评估用户上传的合格文档的内容质量及总下载量，评选出 “既多产、又优质” 的核心贡献者，发予相应奖励：奖项等级奖励内容第 1 名1000 墨值 + 100元内数据库实体书一本第 2 名500墨值 + 爱国者U盘（128GB 双接口）第 3-5 名笔记本电脑支架（可旋转）说明：三类奖项单独评奖、每位用户可重复获得。其中新版本先锋奖数量将根据实际参与情况灵活调整，若参与情况佳则可能增设、反之亦然。五、奖励公布与发放进度公示：为了让大家更加了解自己的参与进度，将在活动期间不定期在活动原文评论区公布最新合格情况。若发现违规行为也欢迎向工作人员反馈，一经核实将取消其参与资格。结果公布：活动结束后 3 个工作日内公布所有获奖名单；奖励发放：墨值将在结果公布后 1-2 个工作日内发放至账户；实物奖励将通过私信收集地址，10 个工作日内寄出。常见问题（Q&A）Q1：如何让我的安装文档下载率更高？：A：建议在标题、简介中写明 “适用版本 + 核心亮点”，若上传的是.zip 等无法预览的格式，可在评论区附关键步骤截图或内容大纲，帮助他人判断价值。Q2：墨值可以干什么？A：墨值是墨天轮社区的“通用货币”，可以用来下载付费文档、赞赏他人文章或进行提问赞赏、直播打赏、兑换商品、参与墨值拍卖等。具体可查看使用说明：https://www.modb.pro/db/446902。Q3：我的文档会被展示在哪里？A：文档上传成功后可在您的控制台、个人主页找到相应链接。管理员也会择优推荐到网站首页及文档页面，为您的文档增加更多曝光。若您觉得您的文档优秀，欢迎您转发分享或自荐给工作人员，我们将为您争取首页推荐、转发等更多曝光。无论是你是初学者还是资深开发者，欢迎你加入本次文档悬赏活动，分享你的优质文档，与更多朋友一起学习进步！未来我们也将针对更多技术主题推出悬赏活动，如果你有推荐的主题也不妨告诉我们！乐知乐享、共同成长！点击跳转上传文档：https://www.modb.pro/docUpload往期活动导航【文档悬赏令】第1号：数据库新版本的安装实操欲了解更多可浏览墨天轮社区，围绕数据人的学习成长提供一站式的全面服务，打造集新闻资讯、在线问答、活动直播、在线课程、文档阅览、资源下载、知识分享及在线运维为一体的统一平台，持续促进数据领域的知识传播和技术创新。关注官方公众号： 墨天轮、 墨天轮平台、墨天轮成长营、数据库国产化 、数据库资讯
作者：马金友， 一名给 MySQL 找 bug 的初级 DBA。爱可生开源社区出品，原创内容未经授权不得随意使用，转载请联系小编并注明来源。本文约 1500 字，预计阅读需要 5 分钟。如果你注意到在 MySQL 中 ORDER BY DESC 查询比 ORDER BY ASC 稍微慢一些，不用担心 —— 这是已知且符合预期的行为。这是因为 InnoDB 的设计和优化是为了进行正向扫描，它使用单向链表结构来组织页面上的记录。因此，向前移动（ASC）的时间复杂度是 O(1)，而向后移动（DESC）的时间复杂度是 O(n)。这篇博客将从存储层面的角度演示这两种算法。1. InnoDB 页面结构1.1 单向链表InnoDB 使用单向链表来组织record。 每个页面有两个虚拟record：infimum 和 supremum，它们分别作为链表的头部和尾部。一旦数据页面包含用户记录，链表就会按逻辑顺序显示。infimum -> rec1 -> rec2 -> rec3 -> rec4 -> ... -> supremum1.2 REC_NEXT每条记录在记录头中额外占用 2 个字节(byte)来存储指向下一条记录的偏移量。constexpr uint32_t REC_NEXT = 2; constexpr uint32_t REC_NEXT_MASK = 0xFFFFUL;例如，infimum 记录的 REC_NEXT 值是 0x00, 0x0d。/** The page infimum and supremum of an empty page in ROW_FORMAT=COMPACT */ static const byte infimum_supremum_compact[] = { /* the infimum record */ 0x01 /*n_owned=1*/, 0x00, 0x02 /* heap_no=0, REC_STATUS_INFIMUM */, 0x00, 0x0d /* pointer to supremum */, 'i', 'n', 'f', 'i', 'm', 'u', 'm', 0, /* the supremum record */ 0x01 /*n_owned=1*/, 0x00, 0x0b /* heap_no=1, REC_STATUS_SUPREMUM */, 0x00, 0x00 /* end of record list */, 's', 'u', 'p', 'r', 'e', 'm', 'u', 'm'};通过 infimum 记录偏移 0x000d，可以得到 supremum 记录。In [1]: infimum_supremum_compact = [ ...: 0x01 , 0x00, 0x02 , 0x00, ...: 0x0d , 'i', 'n', 'f', 'i', 'm', 'u', 'm', 0, ...: 0x01 , 0x00, 0x0b, 0x00, ...: 0x00, 's', 'u', 'p', 'r', 'e', 'm', 'u', 'm' ...: ] ...: In [2]: infimum_supremum_compact[5] Out[2]: 'i' In [3]: infimum_supremum_compact[5+0x000d] Out[3]: 's'1.3 页面目录 (Page Directory)由于单向链表的数据结构，InnoDB 必须扫描整个链表才能找到一条 record，这效率很低。InnoDB 在每个数据页的末尾维护一个动态数组（page directory），数组中的每个元素（槽/slot）存储一条record的位置。/* We define a slot in the page directory as two bytes */ constexpr uint32_t PAGE_DIR_SLOT_SIZE = 2;它不是存储每条记录的地址，而是每个槽指向该槽所管理记录中的最后一条记录。一个槽通常管理 4 到 8 条记录。/* The maximum and minimum number of records owned by a directory slot. The number may drop below the minimum in the first and the last slot in the directory. */ constexpr uint32_t PAGE_DIR_SLOT_MAX_N_OWNED = 8; constexpr uint32_t PAGE_DIR_SLOT_MIN_N_OWNED = 4;第一个槽总是指向 infimum，最后一个槽总是指向 supremum。1.4 N_OWNED每条记录在记录头中占用 4 个位（bit）来存储 N_OWNED。constexpr uint32_t REC_NEW_N_OWNED = 5; /* This is single byte bit-field */ constexpr uint32_t REC_N_OWNED_MASK = 0xFUL;如果记录是槽中的最后一条记录，它的值就是该槽拥有的记录数。否则，值为 0。2. 示例下图展示了数据页面的布局橙色箭头连接了从 rec0 到 rec23 的 24 条用户记录。灰色箭头指向槽所管理的最后一条记录。槽 0 指向 infimum，它包含 1 条记录。槽 n 指向 supremum，它包含 5 条记录。槽 1 指向 rec3，它包含 4 条记录。3. 算法我们将使用以下逻辑 InnoDB 页面布局来理解这两种扫描算法。3.1 正向扫描 (Forward Scan)从 rec10 找到页面上的下一条记录很容易。读取 REC_NEXT 偏移量field_value = mach_read_from_2(rec - REC_NEXT);获取下一条记录的位置return (ut_align_offset(rec + field_value, UNIV_PAGE_SIZE));3.2 反向扫描 (Backward Scan)从rec10找到页面上的前一条记录会更困难。3.2.1 查找哪个槽管理了当前记录 (page_dir_find_owner_slot)① 扫描从当前记录开始的所有记录，直到 n_owned 不为 0。while (rec_get_n_owned_new(r) == 0) { r = rec_get_next_ptr_const(r, true); ... }它会检查 rec10，然后是 rec11。[rec10] --> [rec11] ^ [rec10] --> [rec11] ^因为 rec11 的 n_owned 是 4，所以会跳转到步骤 1.2。② 检查所有槽，直到找到指向步骤 1.1 中记录 r 的槽。rec_offs_bytes = mach_encode_2(r - page); while (UNIV_LIKELY(*(uint16 *)slot != rec_offs_bytes)) { .... slot += PAGE_DIR_SLOT_SIZE; } return (((ulint)(first_slot - slot)) / PAGE_DIR_SLOT_SIZE);它会从最后一个槽（slot n）开始扫描到 slot 0。[n]...[4][3][2][1][0] ^因为 slot n 指向 supremum（不是 rec11），所以会检查下一个槽（slot 4）。[n]...[4][3][2][1][0] ^因为 slot 4 指向 rec15（不是 rec11），所以会检查下一个槽（slot 3）。[n]...[4][3][2][1][0] ^因为 slot 3 指向 rec11，所以会返回 3。3.2.2 扫描当前slot group 以查找前一条记录① 跳转到前一个槽。 因为 slot 3 只持有slot group的最后一条记录，它无法扫描 slot 3 中的所有记录。slot = page_dir_get_nth_slot(page, slot_no - 1); rec2 = page_dir_slot_get_rec(slt);幸运的是，它可以利用前一个槽组的最后一条记录来扫描当前槽组中的所有record。通过检查 slot 2，它会找到 rec7。② 扫描槽组中的所有记录所有 record 匹配当前 record。while (rec != rec2) { prev_rec = rec2; rec2 = page_rec_get_next_low(rec2, true); } return (prev_rec);它会检查 rec7、rec8、rec9，然后是 rec10，直到找到 rec10 的前一条 record，即 rec9。[rec7] --> [rec8] --> [rec9] --> [rec10] --> [rec11] ^[rec7] --> [rec8] --> [rec9] --> [rec10] --> [rec11] ^[rec7] --> [rec8] --> [rec9] --> [rec10] --> [rec11] ^[rec7] --> [rec8] --> [rec9] --> [rec10] --> [rec11] ^4. 时间复杂度正向扫描是 O(1)，但反向扫描是 O(n)，其中 n 是页面目录中的槽数。5. 基准测试5.1 正向扫描 (Forward scan)mysql > select k from sbtest1 order by k asc limit 9999999, 1; +---------+ | k | +---------+ | 8670945 | +---------+ 1 row in set (1.41 sec) mysql > desc select k from sbtest1 order by k asc limit 9999999, 1\G *************************** 1. row *************************** id: 1 select_type: SIMPLE table: sbtest1 partitions: NULL type: index possible_keys: NULL key: k_1 key_len: 4 ref: NULL rows: 9864216 filtered: 100.00 Extra: Using index 1 row in set, 1 warning (0.00 sec)5.2 反向扫描 (Backward scan)mysql > select k from sbtest1 order by k desc limit 9999999, 1; +---------+ | k | +---------+ | 1184614 | +---------+ 1 row in set (2.01 sec) mysql > desc select k from sbtest1 order by k desc limit 9999999, 1\G *************************** 1. row *************************** id: 1 select_type: SIMPLE table: sbtest1 partitions: NULL type: index possible_keys: NULL key: k_1 key_len: 4 ref: NULL rows: 9864216 filtered: 100.00 Extra: Backward index scan; Using index 1 row in set, 1 warning (0.00 sec)Referencesbtr_pcur_t::move_to_nextbtr_pcur_t::move_to_prev
内存一年疯涨 170%，云账单里的“隐性成本”该算清了
2025 年下半年，存储价格又一次成为行业聚焦点。多家市场机构统计显示，2025 年三季度跟四季度，DRAM 和 NAND 价格一路攀升。根据 Tom's Hardware 披露的数据，2025 年 DRAM 合同价同比上涨幅度高达 171.8%，创下历史新高。此轮上涨跟 AI 数据中心建设拓展、服务器需求集中释放紧密相联，还直接引发企业 IT 基础设施采购成本上升。对于依赖自建数据中心或中小 IDC 的企业来说，这种变化带来的冲击尤为剧烈。硬件采购从一次性预算问题，演变为难以预测的长期成本风险。服务器、SSD 和内存条的价格不再稳定，交付周期也更不确定。企业在扩容时不得不承担高价买入、供货延迟的双重压力。因此，将硬件采购压力转化为按需付费的运营支出，把价格波动风险转移给云服务商，正在成为越来越多企业的选择。但问题并未因此结束。随着业务迁移到云端，企业发现云账单中存储与内存的占比仍在持续上升，即便算力配置并未明显升级，总体成本依旧水涨船高。部分团队开始反思：问题是否仅和数据量增多有关，还是资源使用方式本身就存在不合理的地方？目前，多数云实例依旧按固定的 CPU 与内存配比来交付，诸如 2 核 4GB、4 核 8GB 的规格。早期，这种设计可简化资源管理，推动了云计算普及，但如今业务形态有所改变，企业系统一般得同时支撑多样业务，各业务对于算力、内存的消耗不一样，固定规格愈发难以契合实际需求。这导致企业要么部分资源长期闲置，要么不得不面对业务在高峰阶段出现性能瓶颈的风险。当内存价格进入上行周期，这种规格错配带来的浪费被进一步放大：闲置的不再只是资源本身，而是越来越昂贵的成本。正是基于这样的背景，云基础设施走到新的路径分岔口：是继续就资源本身实施配置，还是转变方向围绕应用需求设计算力供给方式？在近期面向中国区合作伙伴召开的发布会上，华为云对 Flexus 云服务器系列规格及性能进行更新，并且展示了其在各种业务负载下的运行表现。该实例基于华为云首创的柔性算力技术，打破 CPU 与内存的固定绑定关系，使企业能够按真实业务需求配置资源，从源头减少内存浪费，并结合智能调度与应用级加速改善长期运行稳定性与算力资源投入产出比。本文将从行业环境变化与技术实现等层面，剖析这种模式背后的思路，以及它所代表的云服务器演进方向。云服务器，开始不太“合身”了云服务器长期采用固定 CPU 与内存的配比，是工程上的一种取舍考量。早期云平台首先得解决的是规模化交付和稳定调度的问题，采用固定规格利于资源池管理，同样便于容量规划及计费设计。当业务形态呈现相对单一阶段，这样的方式尚可接纳。但究其本质它是从平台管理成本角度设计的，并非从业务负载的角度出发。如今业务已不再是单一模式，电商、内容分发、数据库、缓存、AI 推理在一套系统中同步协同运行，对 CPU 以及内存的需求差别明显，固定规格无法精准对应实际负载，企业只能采用超出实际所需的实例型号。云服务器规格跟应用需求普遍不匹配，用户往往被迫去为用不到的算力和内存付费，引发大量资源的闲置浪费。资源浪费只不过是表象罢了，更深层的问题体现为性能优化的复杂度。现实的业务部署不仅涉及操作系统选定，还包含网络参数、系统参数以及应用配置参数。数量往往达到数千级别，缺少专家经验积累，难以达成稳定的最优配置。单是内核跟应用层的参数组合，就已超出普通团队可控范围，调优所用的周期漫长，效果也难以把控。从较长的时间阶段看，云服务器本身一直在不断演变，最初的资源虚拟化阶段，是把物理服务器标准化成可租借的实例；紧接着进入弹性规模阶段，采取自动伸缩的方式去应对流量变化，这两个阶段处理的是存不存在以及是否充足的问题，当下已经迈入第三阶段，关注焦点转向使用是否高效。过去，固定实例曾是工程优势，如今却愈发像是一件穿着不合身的衣服。柔性算力：从“卖规格”到“卖能力”怎样让资源本身更贴近应用？在 Flexus 云服务器 X 实例产品的设计里，华为云引入了柔性算力这一概念。在 Flexus X 实例里，柔性算力首先体现在规格形态的调整变化上。传统实例一般仅仅可在少量固定比例中选择 CPU 跟内存配置，而该实例支持按业务需求实施更精细的组合配置。发布会现场提到，所有 X 实例均支持多种非常规的 CPU/ 内存配比，包括 3:1、2:5、3:7 等组合。这可减少由规格不一致引起的资源闲置，让用户更接近按实际负载付费。然而规格数量增加，并非表示问题自动就解决了，其关键是系统如何判断哪种配置更合适。传统调度大多依据节点上剩余的 CPU 与内存。新方式需要领会业务负载本身，涵盖资源使用结构，以及随时间的变化趋势。Flexus X 实例本质上不再是调度 CPU，而是实际的业务场景。就工程实现而言，这种转变依赖底层架构的支撑，Flexus X 实例借助华为云自研的擎天 QingTian 架构和瑶光云脑调度系统得以实现，经由计算、存储和网络资源的解耦操作，提高了资源组合的自由度，也增强了非标准规格运行状态下的稳定性。此外，柔性算力还意味着配置不再是一次性决定，实例运行时会一直对资源使用状况进行评估，系统会判断当前配置跟负载是否相符，进而给出调整建议，而且还支持算力规格热升降的独家能力。从这个层面看，Flexus X 实例的转变不只是规格数量增多，它更像是把算力从提前打包好的商品，变成可持续优化的能力，实现“应用驱动算力”的最优体验。关键应用加速：算力之外的第二条性能曲线Flexus X 实例不单单改变了资源形态，还进一步深入应用执行层，解决了算力配置合理系统却依旧不稳定的问题。此次规格升级，华为云为数据库以及中间件类的负载引入专属应用级加速机制。Flexus X 实例针对 PostgreSQL、Memcached、MySQL、Redis、Nginx 提供了独立的一键加速能力，由 X-Turbo 应用加速引擎统一驱动。此类优化不会对用户的使用途径做出改变，实例创建结束之后即可启用，平台会把调优工作完成，用户无需插手复杂参数的配置。发布会现场，华为云对该能力实测演示，在 PostgreSQL 的使用场景下，Flexus X 实例的吞吐量达到 2.1 万 + TPS，大概为同规格业界旗舰型实例的 3.4 倍。就数据库这类系统而言，峰值性能仅仅属于一方面，更为关键的是高负载持续状态下的稳定输出能力。业务系统更易受诸如延迟抖动、连接堆积等问题的干扰，而不是单次压测形成的成绩。X-Turbo 的设计目标之一正是实现性能优化长期运行状态下的吞吐与响应稳定性。跟应用级优化同步进行的是，实例规模的进一步扩展。新一代 Flexus X2e 实例的 x86 规格从原本的 32U128G 提升至 64U256G，多核算力提升了约 30%；新增 Flexus KX1 鲲鹏实例，最高可达 80U320G，以覆盖大数据处理、内存数据库这类资源密集型场景。这意味着应用加速机制不再受中小规格环境约束，能在规模更大的资源池里发挥作用。这一系列的变化显示出云服务器性能边界正在转移。过去，性能更多由 CPU 规格和内存容量决定。而如今，应用执行路径、参数组合的方法及调度策略成为同等要紧的变量，在固定规格的时代里，这些优化由用户自己承担，而于 Flexus X 实例中，它们被纳入到算力交付范畴，正是从这一意义出发，云服务器竞争不再只是资源规模大小的比拼，而是发展为聚焦运行效率的系统工程。从工程能力到真实落地：柔性算力如何进入生产系统一项新的算力供给方式，能否切实进入生产系统，首要取决于它是否具备充足的稳定性与可用性。Flexus X 实例可靠性设计向华为云旗舰级云服务器标准看齐，实现单 AZ 99.975% 的可用水平，还有跨 AZ 99.995% 的可用性。这暗示柔性算力没有以牺牲稳定性为交换代价，而是可直接承受核心业务负载的基础设施形态。除了稳定性这一点，规模化使用还取决于运维体系自身是否具有确定性，Flexus X 实例在华为云既有的 SRE 运维体系框架内运行，强调借助标准化变更、容量预测与故障演练减少系统行为的不确定性，实现大规模实例并发运行的可控性。从行业落地的实际来看，柔性算力最先进入的并非那种单一业务场景，而是负载结构繁杂、资源使用波动大的系统类型。其已经在医疗电商平台迁移、连锁零售系统、医药行业信息化平台、游戏服务器迁移等场景大规模部署，用以承载数据库、中间件及核心交易服务。中软国际智能集团云业务部副总经理王春玉在发布会上分享，团队为某大型生物医药集团搭建系统的时候，引入 Flexus X 实例作为数据库及业务服务的主要承载环境，在原有系统架构未改变的情形下完成迁移，而且在性能满足要求的前提下，达成约 30% 的综合成本下降。王春玉还谈到，其团队服务的一家专业酒水直营连锁品牌，把部分核心业务迁移到 Flexus X 实例而后，通过规格按需匹配与资源利用率优化，实现整体云资源成本约 15% 的下降。这些亮眼的结果主要源于两方面：一是实例规格跟业务负载的匹配度有所提升，降低了长期闲置资源的数量；二是借助应用级加速与调度优化，降低了单位业务量所需的算力规模。从这些真实的实际部署案例能看出，Flexus X 实例的用户一般有几个共同特性：业务负载呈现明显波动，系统结构相对复杂，然而运维及架构团队的规模较为有限，同时对长期云资源的成本敏感度较高。Flexus X 实例在未对业务形态本身作出改变的情况下，却降低了基础设施对业务扩展所施加的约束强度，让按照业务形态去配置算力成为可践行的工程实践。可以预见，未来企业买的不再是服务器，而是业务效率。Flexus X 实例凸显了云服务器设计思路的一次转向：由“卖规格”过渡到“交付能力”，从“静态资源”过渡到“智能算力”，在 AI 成为主流计算负载的未来，此种转变大概率不会再是差异化优势，而是云基础设施的必要门槛。
【实操教程】免费外汇 API 获取实时汇率数据（Python+WebSocket 完整实现）
在金融工具开发场景中，实时汇率数据是高频交易、跨境支付工具的核心依赖，但商用数据源成本高、轻量需求下接入性价比低，成了很多开发者的痛点。外汇市场日均 6 万亿美元的交易量，要求数据具备毫秒级时效性，而免费外汇 API 恰好能以零成本解决这一问题。本文从开发者视角，拆解基于免费外汇 API 的实时汇率获取方案，附可直接运行的 Python 代码，适配快速落地需求。一、免费外汇 API 的核心优势（开发者视角）对开发者而言，免费外汇 API 的价值远不止 “免费”，更在于适配开发效率：零成本试错：无需支付数据源订阅费，降低工具原型开发阶段的成本，聚焦核心业务逻辑；低集成门槛：接口设计标准化，文档清晰，无需额外适配开发，10 分钟即可完成接入；高性能传输：API 支持 WebSocket 协议，相比 REST API 的 “请求 - 响应” 模式，实现持久化数据通道，数据延迟低至毫秒级，满足实时性需求。二、实时汇率获取实操（完整代码 + 解析）以下为基于 WebSocket 的 EUR/USD 实时汇率获取代码，代码结构清晰，包含完整异常处理，可直接复用至项目中。1.完整代码实现import websocket import json # WebSocket URL，具体API地址根据你选择的API提供商来获取 url = "wss://api.alltick.co/forex/marketdata" # 假设的API URL # 定义请求的参数 params = { "pair": "EURUSD", # 你需要查询的货币对 "apikey": "YOUR_API_KEY" # 替换成你自己的API密钥 } # WebSocket消息格式 def on_open(ws): print("Connection established") # 发送请求数据 ws.send(json.dumps(params)) def on_message(ws, message): # 处理返回的数据 data = json.loads(message) if 'rate' in data: print(f"当前汇率：EUR/USD = {data['rate']}") else: print("没有获取到汇率数据") def on_error(ws, error): print(f"发生错误：{error}") def on_close(ws): print("连接关闭") # 创建WebSocket连接 ws = websocket.WebSocketApp(url, on_open=on_open, on_message=on_message, on_error=on_error, on_close=on_close) # 运行WebSocket连接 ws.run_forever()2.核心逻辑解析协议选型：采用 WebSocket 长连接，避免短连接频繁建联的性能损耗，适配实时数据持续推送的场景，这是相比传统 API 的核心优势；参数设计：仅保留pair（货币对）和apikey（密钥）两个核心参数，极简设计降低使用成本，替换参数即可适配多币种需求；回调函数：connect_open：连接建立后自动发送请求，无需手动触发；receive_message：解析返回数据，提取核心汇率字段，可直接对接业务逻辑；catch_error/close_connection：捕获连接异常、处理连接关闭，保障程序稳定性。三、扩展与优化建议多币种批量获取：构建货币对列表（如["EURUSD", "USDJPY", "GBPUSD"]），循环发起请求，实现多币种数据同步；异常重连机制：在catch_error中添加重连逻辑，避免因网络波动导致数据中断：python 运行 def catch_error(ws, error): print(f"数据连接异常：{error}") ws.close() ws.run_forever() # 自动重连数据持久化：在receive_message中接入数据库（如 MySQL、Redis），将实时汇率数据落地，用于后续分析或策略执行。总结免费外汇 API 是轻量级金融工具开发的最优选择，零成本且能满足实时性需求；WebSocket 协议是实现实时汇率获取的核心技术，相比 REST API 具备低延迟、长连接优势；代码核心在于回调函数的设计，异常处理是保障生产环境稳定运行的关键。如果在接入过程中遇到 API 密钥申请、协议适配等问题，欢迎在评论区交流，共同优化实现方案。
Spring Boot 3 集成 Apache Calcite：多数据源查询的终极解决方案
熟悉 Spring Boot 3 的开发者，都知道它在简化开发流程、提高开发效率方面的出色表现吧！但是，在实际业务场景中，大家肯定都碰到过这样的棘手问题：订单数据存放在 MySQL 里，库存数据在 PostgreSQL 中，用户数据又保存在 MongoDB 中，当多种数据源同时存在时，想要实现统一查询简直比登天还难。所以呢，今天我就亮出我的“终极大招”——Apache Calcite，着重给大家讲讲它怎样与 Spring Boot 3 实现无缝集成，还会分享一些可以直接拿来使用的经典应用场景。掌握了这一招，多数据源查询的难题就能轻松解决啦！一、核心认知：Apache Calcite 为何是多数据源查询的利器？在动手集成前，咱们先把核心逻辑搞明白：为啥 Calcite 能成为多数据源查询的“万能钥匙”？它的核心优势到底在哪？1.1 不止是查询引擎：Calcite 的核心定位Apache Calcite 本质是一个动态数据管理框架，而非传统的数据库。它最核心的价值在于“解耦”——将数据存储与数据查询分离，无论数据存在哪里、是什么格式，都能通过统一的 SQL 接口进行查询。说通俗点，Calcite 就像个“超级数据翻译官”——不管数据藏在哪个数据源里、是什么格式，你只要写一套标准 SQL，它就能翻译成对应数据源能懂的指令，最后把结果整理成统一格式返回。这也是它能搞定多数据源查询的核心秘诀！1.2 Calcite 的核心能力拆解统一 SQL 接口：支持标准 SQL，无论底层是关系型数据库（MySQL、PostgreSQL）、非关系型数据库（MongoDB、Redis），还是文件（CSV、Parquet）、大数据引擎（Hive、Spark），都能通过同一套 SQL 查询。强大的查询优化：内置基于规则和成本的查询优化器，能自动优化 SQL 执行计划，提升查询效率，尤其是在复杂多表关联、跨数据源查询场景下，优化效果明显。灵活的数据源适配：通过“适配器（Adapter）”机制适配不同数据源，社区已提供大量现成适配器，也支持自定义开发，适配特殊数据源。轻量级集成：核心依赖体积小，无复杂依赖，可轻松集成到 Spring Boot、Spring Cloud 等主流 Java 开发框架中，无需单独部署独立服务（也支持独立部署）。二、重点实战：Spring Boot 3 集成 Calcite 核心步骤既然大家都熟悉 Spring Boot 3 的基础操作，我就不啰嗦项目搭建这些常规步骤了，直接聚焦 Calcite 集成的核心环节，每一步都附完整代码和避坑提醒，跟着做就能成！2.1 核心依赖引入第一步先引依赖，在 pom.xml 里加好 Calcite 核心包、对应数据源的适配器，再配上 MyBatis Plus 的核心依赖（替换掉原来的 Jdbc 依赖就行），具体如下： <dependency> <groupId>org.apache.calcite</groupId> <artifactId>calcite-core</artifactId> <version>1.36.0</version> </dependency> <dependency> <groupId>org.apache.calcite</groupId> <artifactId>calcite-mysql</artifactId> <version>1.36.0</version> </dependency> <dependency> <groupId>org.apache.calcite</groupId> <artifactId>calcite-mongodb</artifactId> <version>1.36.0</version> </dependency> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-boot-starter</artifactId> <version>3.5.5</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>druid-spring-boot-starter</artifactId> <version>1.2.20</version> </dependency>这里有 3 个避坑点必须强调下：Calcite 所有组件版本要统一，不然容易出现类加载异常；MyBatis Plus 得选适配 Spring Boot 3 的版本（3.5.3+）；一定要加连接池依赖，不然 Calcite 数据源没法被 MyBatis Plus 正常管理。2.2 核心配置：Calcite 模型文件编写模型文件是 Calcite 识别数据源的关键，一般用 JSON 格式，放在 resources 目录下命名为 calcite-model.json 就行。下面给大家一个适配 MySQL 和 MongoDB 双数据源的示例，直接改改连接信息就能用：{ "version": "1.0", "defaultSchema": "ecommerce", "schemas": [ { "name": "ecommerce", "type": "custom", "factory": "org.apache.calcite.adapter.jdbc.JdbcSchema$Factory", "operand": { "jdbcUrl": "jdbc:mysql://localhost:3306/ecommerce_order?useSSL=false&serverTimezone=UTC", "username": "root", "password": "123456", "driver": "com.mysql.cj.jdbc.Driver" } }, { "name": "user_mongo", "type": "custom", "factory": "org.apache.calcite.adapter.mongodb.MongoSchema$Factory", "operand": { "host": "localhost", "port": 27017, "database": "user_db", "collection": "user_info" } } ] }几个关键配置给大家解释清楚，避免踩坑：defaultSchema：默认查询的 Schema，可省略，查询时需指定 Schema 名称（如 ecommerce.order、user_mongo.user_info）。factory：对应数据源的适配器工厂类，Calcite 已为主流数据源提供现成工厂，自定义数据源需实现自己的 Factory。operand：数据源连接参数，根据数据源类型不同配置不同参数（如 MySQL 的 jdbcUrl、MongoDB 的 host/port）。2.3 Spring Boot 集成 Calcite + MyBatis Plus 核心配置这一步是核心，主要分两步走：配置好 Calcite 数据源；让 MyBatis Plus 用上这个数据源，顺便把 mapper 扫描、分页插件这些基础参数配好。直接上配置类代码：import com.baomidou.mybatisplus.annotation.DbType; import com.baomidou.mybatisplus.autoconfigure.ConfigurationCustomizer; import com.baomidou.mybatisplus.extension.plugins.MybatisPlusInterceptor; import com.baomidou.mybatisplus.extension.plugins.inner.PaginationInnerInterceptor; import org.apache.calcite.jdbc.CalciteConnection; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.io.support.PathMatchingResourcePatternResolver; import javax.sql.DataSource; import java.sql.Connection; import java.sql.DriverManager; import java.util.Properties; @Configuration // MyBatis Plus mapper 接口扫描（指定 mapper 包路径） @MapperScan(basePackages = "com.example.calcite.mapper") public class CalciteMybatisPlusConfig { // 1. 配置 Calcite 数据源（核心，与原逻辑一致） @Bean public DataSource calciteDataSource() throws Exception { Properties props = new Properties(); props.setProperty("model", "classpath:calcite-model.json"); Connection connection = DriverManager.getConnection("jdbc:calcite:", props); CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class); return calciteConnection.getDataSource(); } // 2. 配置 MyBatis Plus 的 SqlSessionFactory，指定使用 Calcite 数据源 @Bean public SqlSessionFactory sqlSessionFactory(DataSource calciteDataSource) throws Exception { MybatisSqlSessionFactoryBean sessionFactory = new MybatisSqlSessionFactoryBean(); // 注入 Calcite 数据源 sessionFactory.setDataSource(calciteDataSource); // 配置 mapper.xml 文件路径（如果使用 XML 方式编写 SQL） sessionFactory.setMapperLocations(new PathMatchingResourcePatternResolver() .getResources("classpath:mapper/*.xml")); // 配置 MyBatis Plus 全局参数（可选） org.apache.ibatis.session.Configuration configuration = new org.apache.ibatis.session.Configuration(); configuration.setMapUnderscoreToCamelCase(true); // 下划线转驼峰 sessionFactory.setConfiguration(configuration); // 注入 MyBatis Plus 插件（如分页插件） sessionFactory.setPlugins(mybatisPlusInterceptor()); return sessionFactory.getObject(); } // 3. MyBatis Plus 分页插件（可选，复杂查询分页用） @Bean public MybatisPlusInterceptor mybatisPlusInterceptor() { MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor(); interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.MYSQL)); // 适配 Calcite 兼容的 MySQL 语法 return interceptor; } // 4. 配置事务管理器（可选，需要事务支持时添加） @Bean public PlatformTransactionManager transactionManager(DataSource calciteDataSource) { return new DataSourceTransactionManager(calciteDataSource); } }核心逻辑给大家捋一捋：先通过 Calcite 创建统一的数据源，再把它注入到 MyBatis Plus 的 SqlSessionFactory 里。这样一来，咱们后续写代码就完全是 MyBatis Plus 的熟悉风格了，不管是 Mapper 接口还是 XML 映射文件，都能直接用，跨数据源查询的复杂逻辑全交给 Calcite 处理。2.4 核心查询实现（MyBatis Plus 风格）接下来就是大家最熟悉的查询实现环节了，我用 MyBatis Plus 最常用的“Mapper 接口+注解”和“XML”两种方式来演示，还是以 MySQL 订单表和 MongoDB 用户表的关联查询为例，大家可以根据自己的习惯选：(1)定义实体类（对应跨数据源查询结果，可使用 lombok 简化代码）import lombok.Data; @Data public class UserOrderVO { private String orderId; // 订单 ID（来自 MySQL） private String orderTime; // 下单时间（来自 MySQL） private BigDecimal amount; // 订单金额（来自 MySQL） private String userName; // 用户名（来自 MongoDB） private String phone; // 手机号（来自 MongoDB） private String userId; // 用户 ID（关联字段） }(2)定义 Mapper 接口（MyBatis Plus 风格，无需编写实现类）import com.baomidou.mybatisplus.core.mapper.BaseMapper; import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Select; import java.util.List; // 继承 BaseMapper，获得 MyBatis Plus 基础 CRUD 能力 public interface UserOrderMapper extends BaseMapper<UserOrderVO> { // 注解方式编写跨数据源关联 SQL @Select("SELECT " + "o.order_id AS orderId, o.order_time AS orderTime, o.amount, " + "u.user_name AS userName, u.phone, o.user_id AS userId " + "FROM ecommerce.order o " + // ecommerce：MySQL 的 Schema；order：订单表 "JOIN user_mongo.user_info u " + // user_mongo：MongoDB 的 Schema；user_info：用户表 "ON o.user_id = u.user_id " + "WHERE o.user_id = #{userId}") List<UserOrderVO> queryUserOrderByUserId(@Param("userId") String userId); }(3)编写 Service 层import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import org.springframework.stereotype.Service; import java.util.List; @Service public class UserOrderServiceImpl extends ServiceImpl<UserOrderMapper, UserOrderVO> implements UserOrderService { @Override public List<UserOrderVO> getUserOrderByUserId(String userId) { // 调用 Mapper 接口方法，实现跨数据源查询 return baseMapper.queryUserOrderByUserId(userId); // 若使用 XML 方式：return baseMapper.queryUserOrderByUserIdWithXml(userId); } }(4)编写 Controller 层import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RestController; import java.util.List; @RestController public class CrossDataSourceQueryController { @Autowired private UserOrderService userOrderService; @GetMapping("/user/order/{userId}") public List<UserOrderVO> queryUserOrder(@PathVariable String userId) { // 调用 Service 方法，返回跨数据源查询结果 return userOrderService.getUserOrderByUserId(userId); } }最后再划 3 个重点，确保大家少走弯路：实体类字段要和查询结果列名对应，用别名适配下划线转驼峰更省心；Mapper 接口继承 BaseMapper 后，MyBatis Plus 的分页、条件构造器这些功能都能直接用，复杂查询也能轻松搞定；咱们写的都是标准 SQL，Calcite 会自动解析适配不同数据源，完全不影响大家原来的开发习惯。三、深度解析：Calcite 的经典使用场景讲完了集成步骤，再跟大家深度拆解下 Calcite 的经典落地场景。毕竟技术最终要服务于业务，这些场景都是我在实际项目中常用到的，拿来就能用！第一个经典场景是多系统数据融合查询，这也是企业级中台的核心需求。做企业级中台的小伙伴肯定深有体会，大型企业里数据都是分散的——订单系统用 MySQL，用户系统用 MongoDB 存行为数据，库存系统用 PostgreSQL。要是想做“用户-订单-库存”全链路分析，传统做法得分别调三个系统的接口，再在业务层手动整合数据，不仅效率低，还容易出错。用 Calcite 分别适配这三个数据源后，只要写一套标准 SQL 就能实现跨数据源关联查询，咱们用 Spring Boot 3 搭好接口服务，业务层完全不用管数据存在哪，专注核心业务逻辑就行，亲测开发效率能提升 50%以上，再也不用写重复的接口调用和数据整合代码，而且 Calcite 的查询优化器会自动优化关联逻辑，查询效率也能跟上。第二个场景是实时数据与离线数据联动查询，做电商的小伙伴应该经常遇到这类需求。比如实时订单数据存在 Kafka 里，历史订单数据存在 Hive 里，运营需要实时查看“今日订单+近 30 天历史订单”的汇总数据来做实时监控和决策。这种情况不用麻烦地把 Kafka 数据同步到 Hive，也不用把 Hive 数据同步到实时库，直接用 Calcite 的 Kafka 适配器（calcite-kafka）和 Hive 适配器（calcite-hive），就能把实时流数据和离线数据放到同一个查询体系里，写一条 SQL 就能实现“实时+离线”数据的联合查询，既省了大量数据同步成本，又能兼顾实时性和准确性，还支持增量查询。第三个场景是自定义数据源适配，主要解决特殊格式数据查询的难题。企业里总有很多 CSV、Excel、Parquet 格式的文件数据，传统做法是先把这些文件导入数据库才能查询，步骤又多又耗时，尤其是临时做数据分析的时候，导入数据库的成本太高了。而 Calcite 内置了文件适配器（calcite-file），支持直接查询这些文件数据，根本不用导入数据库。咱们再结合 Spring Boot 3 的文件上传功能，还能实现“文件上传后直接用 SQL 查询”的需求，临时分析数据超方便。如果有企业内部的特殊格式文件，比如自定义的二进制文件，也可以自己实现 Calcite 的 SchemaFactory 和 TableFactory 接口，写个自定义适配器，就能适配这些特殊数据源了。四、避坑指南：集成注意事项与优化建议4.1 这些坑一定要避开！适配器版本要统一：Calcite 核心依赖和各数据源适配器的版本必须一致，不然很容易出现类加载异常，这个坑我踩过，大家一定要注意。模型文件配置要规范：Schema 名称、表名要清晰，别重复；数据源的地址、端口、账号密码这些连接参数一定要准确，错一个就会连接失败。要考虑数据源性能：跨数据源查询的性能取决于最慢的那个数据源，所以要确保每个数据源自身性能没问题，不然会拖慢整个查询。4.2 优化小技巧，查询更快更稳启用 Calcite 缓存：配置一下 Calcite 的元数据缓存和查询计划缓存，能减少重复解析和元数据查询的时间，提升查询效率。优化 SQL 写法：尽量避免复杂的多表关联，能把过滤条件下推到数据源的就尽量下推。虽然 Calcite 会自动优化，但手动优化后的效果会更好。自定义优化规则：如果是特别复杂的业务场景，可以自己实现 Calcite 的 OptimizerRule 接口，写自定义的查询优化规则，进一步提升查询效率。五、本文总结最后总结一下，对于熟悉 Spring Boot 3 的咱们来说，集成 Calcite 的关键就是理解它“统一查询”的核心思想，把模型文件写对、核心 Bean 配置好，就能快速实现多数据源查询能力了。https://mybj123.com/28732.html