这两年数据库圈有点像3年前的云原生圈："分布式"、"新一代内核"、"重构存储引擎"这些词突然又密集起来了。

前几天刷群，看到有人转了 OpenTeleDB 的开源消息，说是"基于 PostgreSQL 的新一代内核"。说实话，我第一反应是：又一个魔改 PG？

但看到里面提到一个点：原位更新 + Undo 引擎（XStore），我还是没忍住下了源码。 因为这恰好戳中我这些年被 PG 折磨得最狠的痛点：

表膨胀、autovacuum 抽风、性能像心电图一样忽高忽低。

所以这次我没看 PPT，也没看宣传稿，直接跑到机器上拆了半天，想看看它究竟动了 PG 的哪根"老筋"。

一、先说结论：XStore 不是快，而是"稳"

我装的是 OpenTeleDB 的 17.6 内核版。 创建方式很直观：

SELECT relname, amname
FROM pg_class c
JOIN pg_am a ON c.relam = a.oid
WHERE relname = 'test_xstore';

这一步其实就已经很有意思了------它不是 fork 了一套新引擎，而是作为插件挂进去的。 这个思路我很认可：

• 不绑死 PG 版本
• 能跟着大版本升级
• 出问题可以随时回退

像 Citus、openHalo 这些"成功插件化路线"的项目，本质都是这个思路。

二、打开数据目录，我第一次意识到：它真不是换皮

在 $PGDATA 下面，多了一个非常显眼的目录：

drwx------ 2 postgres postgres  4096 Nov  3 20:15 undo

这就是 XStore 的核心： 它不是靠多版本链来维护 MVCC，而是靠 Undo 日志回滚。

这点和 Oracle、MySQL InnoDB 的逻辑更像。

也正是它敢说"原位更新"的底气来源。

三、插入测试：它不快，但很"诚实"

我用同样的参数，在同一台机器上跑了两组：

INSERT INTO test_xstore (name, value)
SELECT md5(random()::text), (random()*1000)::int
FROM generate_series(1,10000000);

INSERT INTO test_heap (name, value)
SELECT md5(random()::text), (random()*1000)::int
FROM generate_series(1,10000000);

结果是：

写慢了将近一倍。这点我反而觉得真实：因为 XStore 在写数据页的同时，还要写一份 Undo。物理写入翻倍，吞吐下降是必然的。如果一个系统告诉你"原位更新 + Undo 还更快"，那我反而会不太信。

四、创新实验：模拟1千万数据的存储膨胀对比

我设计了一项创新实验：在 1000 万条级别的大数据量下，评估 XStore 与 Heap 表在高频更新下的空间膨胀、索引稳定性以及查询性能表现。该实验主要有两个创新点：

大规模数据模拟

1. 使用 generate_series(1,10000000) 生成 1000 万条数据，保证数据量级对存储膨胀影响明显。
2. 初始数据包括 id、name、value 和 updated_at 四列，与前期实验一致，但数据量增加十倍，以模拟真实大规模 OLTP 系统负载。

多维度空间分析

1. 不仅监控表总大小，还分别统计索引占用和 TOAST 表空间。
2. 每轮更新后，通过 pg_relation_size、pg_total_relation_size 和 pg_indexes_size 获取精细化指标。
3. 引入 可视化趋势分析，绘制表空间增长曲线，以直观展示 XStore 与 Heap 的差异。

4.1 实验设计

表结构

CREATE TABLE xstore_large (
    id SERIAL PRIMARY KEY,
    name TEXT,
    value INT,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) USING XSTORE;

CREATE TABLE heap_large (
    id SERIAL PRIMARY KEY,
    name TEXT,
    value INT,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

初始化 1000 万条数据

INSERT INTO xstore_large (name, value)
SELECT 'name_' || g, g
FROM generate_series(1, 10000000) AS g;

INSERT INTO heap_large (name, value)
SELECT 'name_' || g, g
FROM generate_series(1, 10000000) AS g;

先对现在存入1000w数据的空间监控与记录一下如下。

SELECT
    pg_size_pretty(pg_total_relation_size('xstore_large')) AS xstore_total,
    pg_size_pretty(pg_indexes_size('xstore_large')) AS xstore_index,
    pg_size_pretty(pg_total_relation_size('heap_large')) AS heap_total,
    pg_size_pretty(pg_indexes_size('heap_large')) AS heap_index;

 xstore_total | xstore_index | heap_total | heap_index
--------------+--------------+------------+------------
 985 MB       | 388 MB       | 789 MB     | 214 MB

多轮全表更新

• 连续 5 轮更新，每轮更新 value 和 updated_at，模拟写入密集场景：

UPDATE xstore_large
SET value = value + 1, updated_at = CURRENT_TIMESTAMP;

UPDATE heap_large
SET value = value + 1, updated_at = CURRENT_TIMESTAMP;

空间监控与记录

SELECT
    pg_size_pretty(pg_total_relation_size('xstore_large')) AS xstore_total,
    pg_size_pretty(pg_indexes_size('xstore_large')) AS xstore_index,
    pg_size_pretty(pg_total_relation_size('heap_large')) AS heap_total,
    pg_size_pretty(pg_indexes_size('heap_large')) AS heap_index;

第一轮：

 xstore_total | xstore_index | heap_total | heap_index
--------------+--------------+------------+------------
 985 MB       | 388 MB       | 1578 MB    | 428 MB

第五轮：

 xstore_total | xstore_index | heap_total | heap_index
--------------+--------------+------------+------------
 985 MB       | 388 MB       | 1628 MB    | 428 MB

4.2 千万数据更新膨胀可视化

五、实验结论

这组 1000 万级数据 + 多轮全表更新的实验，其实把 PG 传统 Heap 表的"老问题"放大得非常清楚。

最核心的对比结果只有一句话：

XStore 的空间是线性的、可预测的；Heap 表的空间是失控的、不可预测的。

具体来看：

1. 表空间膨胀

a. Heap 表在第一次更新后，表体空间直接翻倍，从 789MB 飙到 1578MB。

b. 之后每一轮更新，虽然增长幅度趋缓，但空间再也回不到初始状态。

c. XStore 从头到尾不变： 985MB → 985MB → 985MB

2. 索引体积稳定性

a. Heap 表索引从 214MB 膨胀到 428MB，且在后续更新中保持"高位横盘"。

b. XStore 的索引尺寸始终维持在 388MB 左右，没有明显漂移。

3. 更新行为本质差异

a. Heap：每一次 UPDATE，本质都是 DELETE + INSERT → 老版本残留 → 表膨胀 → 索引碎片 → autovacuum 压力。

b. XStore：真正的原位更新 → 历史版本进 Undo → 主表物理页不变 → 无膨胀。

4. 长期可运维性

a. 在 Heap 表上，如果你不 VACUUM，它一定会慢； 如果你 VACUUM，系统一定会抖。

b. 在 XStore 上，这两件事都不再是必选项。

这意味着什么？

它不是让你飞起来，而是让你不再塌方。

六、我的心得

说实话，这几年我已经对"新一代数据库内核"这类说法有点免疫了。大多数项目，要么是在 PG 上糊一层分布式壳； 要么就是换个名字，重新卖一遍 MVCC。而 XStore 给我的感觉不一样。它没有试图掩盖代价。写入更慢， IO 更多，架构更复杂。

但它正面承认了一个事实：

PostgreSQL 的 MVCC，在高频更新场景下已经接近物理极限。

这不是参数调优能解决的事，也不是加机器能扛住的事，而是存储模型本身的问题。这些年我见过太多系统：白天 QPS 很稳，半夜 autovacuum 开始清垃圾，延迟突然拉长，业务报警，DBA 开始手工 VACUUM / REINDEX / CLUSTER，第二天继续循环。

这不是运维水平的问题，而是模型在和现实硬扛。XStore 让我第一次意识到：原来 PG 也可以选择不走这条老路。它没有追求"更快"，而是选择了一个更难、但更稳的方向：

• 用 Undo 换空间可控
• 用写放大换性能平滑
• 用工程复杂度换系统长期可预期性

如果你是写多、更新密集型 OLTP 系统，如果你被表膨胀、索引碎片、autovacuum 抽风折磨过，那你会和我一样---不一定立刻用它，但你会开始认真看它。这大概就是我这次拆源码、跑实验，最大的收获。

使用托管数据库部署 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 能够在带宽受限的环境中实现高效、实时的消息传递。

关于 Coreflux

Coreflux 提供了一个轻量级 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 代理 之间的安全通信，无需公开访问。

1. 登录你的 DigitalOcean 控制面板
2. 从左侧导航栏进入 网络 → VPC
3. 点击 创建 VPC 网络

1. 为物联网自动化配置你的 VPC：

   • 名称：coreflux-integrations-vpc（或你的 VPC 名称）
   • 数据中心区域：选择法兰克福（或你首选的区域）
   • IP 范围：使用默认值或根据需要配置
   • 描述：为你的 MQTT 代理和数据库 网络添加有意义的描述
2. 点击 创建 VPC 网络

VPC 将为你所有的物联网资源提供隔离的网络，确保 Coreflux MQTT 代理 和 托管数据库 之间的安全通信。有关 VPC 配置的更多详细信息，请参阅我们关于创建 VPC 网络的教程。

步骤 2 — 为可扩展存储设置托管数据库

根据你的物联网应用需求，选择以下数据库选项之一：

• PostgreSQL：适用于需要关系查询、ACID 合规性和复杂关系的结构化数据
• MySQL：适用于结构化工作负载和具有强一致性及广泛工具支持的事务性查询
• MongoDB：适用于具有可变模式的灵活文档存储和快速开发
• OpenSearch：适用于高级搜索、分析、日志分析和时间序列数据可视化

设置 PostgreSQL 托管数据库

当你的物联网工作负载需要关系模式、强一致性 和 高级 SQL 分析，并由自动备份、监控和维护支持时，DigitalOcean 上的托管 PostgreSQL 是一个很好的选择。

1. 从 DigitalOcean 控制面板，导航到 数据库
2. 点击 创建数据库集群

3. 为物联网自动化配置你的 PostgreSQL 集群：

   • 数据库引擎：选择 PostgreSQL
   • 版本：选择最新的稳定版本
   • 数据中心区域：选择法兰克福（与你的 VPC 相同）
   • VPC 网络：选择你创建的 coreflux-integrations-vpc
   • 数据库集群名称：postgresql-coreflux-test
   • 项目：选择你的目标项目

4. 根据你的 物联网 需求选择你的计划：

   1. 对于开发：基础 计划，1 GB RAM
   2. 对于生产：通用型 或更高，用于可扩展存储
5. 点击 创建数据库集群

托管数据库 创建过程通常需要 1-5 分钟。完成后，你将被重定向到数据库概览页面，在那里你可以查看连接详细信息并执行管理操作。

为 MQTT 代理集成配置 PostgreSQL 数据库访问

系统将提示你进行入门步骤，显示你的连接详细信息，你可以配置入站访问规则（建议限制为你的 IP 和仅 VPC）。

1. 点击 开始使用 来配置你的 PostgreSQL 数据库

2. （可选操作）限制入站连接：

   • 添加你本地计算机的 IP 以进行管理访问
   • droplet 将通过 VPC 网络自动获得允许

对于连接详细信息，你将看到两个选项 - 公共网络和 VPC 网络。第一个用于像 DBeaver 这样的工具进行外部访问，而第二个将由 Coreflux 服务用于访问数据库。

1. 记下提供的连接详细信息，包括公共访问和 VPC 访问（每种都有不同的详细信息）：

   • 主机：你的数据库主机名
   • 用户：默认管理员用户
   • 密码：自动生成的安全密码
   • 数据库：身份验证数据库名称

测试 PostgreSQL 数据库连接

你可以使用提供的连接参数，使用公共访问凭证通过 DBeaver 测试 PostgreSQL 连接：

创建 PostgreSQL 应用程序数据库和用户（可选）

为了更好的安全性和组织性，为你的 物联网自动化 应用程序创建一个专用用户和数据库。这也可以通过 DBeaver 或 CLI 完成，但 DigitalOcean 提供了一种用户友好的方法：

1. 转到你的 托管数据库 集群中的 用户与数据库 选项卡

2. 创建用户：

   • 用户名：coreflux-broker-client
   • 密码：自动生成

3. 创建数据库：

   • 数据库名称：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 是结构化、事务性物联网数据的理想选择。

1. 从 DigitalOcean 控制面板，导航到 数据库
2. 点击 创建数据库集群

3. 为物联网自动化配置你的 MySQL 集群：

   • 数据库引擎：选择 MySQL
   • 版本：选择最新的稳定版本
   • 数据中心区域：选择法兰克福（与你的 VPC 相同）
   • VPC 网络：选择你创建的 coreflux-integrations-vpc
   • 数据库集群名称：mysql-coreflux-test
   • 项目：选择你的目标项目

4. 根据你的 物联网 需求选择你的计划：

   • 对于开发：基础 计划，1 GB RAM
   • 对于生产：通用型 或更高，用于可扩展存储
5. 点击 创建数据库集群

托管数据库 创建过程通常需要 1-5 分钟。完成后，你将被重定向到数据库概览页面，在那里你可以查看连接详细信息并执行管理操作。

为 MQTT 代理集成配置 MySQL 数据库访问

系统将提示你进行入门步骤，显示你的连接详细信息，你可以配置入站访问规则（建议限制为你的 IP 和仅 VPC）。

1. 点击 开始使用 来配置你的 MySQL 数据库

2. （可选操作）限制入站连接：

   • 添加你本地计算机的 IP 以进行管理访问
   • droplet 将通过 VPC 网络自动获得允许

对于连接详细信息，你将看到两个选项 - 公共网络和 VPC 网络。第一个用于像 DBeaver 这样的工具进行外部访问，而第二个将由 Coreflux 服务用于访问数据库。

1. 记下提供的连接详细信息，包括公共访问和 VPC 访问（每种都有不同的详细信息）：

   • 主机：你的数据库主机名
   • 用户：默认管理员用户
   • 密码：自动生成的安全密码
   • 数据库：身份验证数据库名称

测试 MySQL 数据库连接

你可以使用提供的连接参数，使用公共访问凭证通过 DBeaver 测试 MySQL 连接。

注意： 你可能需要更改 DBeaver 的驱动程序设置——设置 allowPublicKeyRetrieval = true。

创建 MySQL 应用程序数据库和用户（可选）

为了更好的安全性和组织性，为你的 物联网自动化 应用程序创建一个专用用户和数据库。这也可以通过 DBeaver 或 CLI 完成，但 DigitalOcean 提供了一种用户友好的方法：

1. 转到你的 托管数据库 集群中的 用户与数据库 选项卡

2. 创建用户：

   • 用户名：coreflux-broker-client
   • 密码：自动生成

3. 创建数据库：

   • 数据库名称：coreflux-broker-data

设置 MongoDB 托管数据库

托管 MongoDB 非常适合灵活或不断演变的物联网负载，让你能够存储异构的传感器文档，而无需严格模式，同时平台处理复制、备份和监控。

1. 从 DigitalOcean 控制面板，导航到 数据库
2. 点击 创建数据库集群

3. 为物联网自动化配置你的 MongoDB 集群：

   • 数据库引擎：选择 MongoDB
   • 版本：选择最新的稳定版本
   • 数据中心区域：选择法兰克福（与你的 VPC 相同）
   • VPC 网络：选择你创建的 coreflux-integrations-vpc
   • 数据库集群名称：mongodb-coreflux-test
   • 项目：选择你的目标项目

4. 根据你的 物联网 需求选择你的计划：

   • 对于开发：基础 计划，1 GB RAM
   • 对于生产：通用型 或更高，用于可扩展存储
5. 点击 创建数据库集群

托管数据库 创建过程通常需要 1-5 分钟。完成后，你将被重定向到数据库概览页面，在那里你可以查看连接详细信息并执行管理操作。

为 MQTT 代理集成配置 MongoDB 数据库访问

系统将提示你进行入门步骤，显示你的连接详细信息，你可以配置入站访问规则（建议限制为你的 IP 和仅 VPC）。

1. 点击 开始使用 来配置你的 MongoDB 数据库

2. （可选）限制入站连接：

   • 添加你本地计算机的 IP 以进行管理访问
   • droplet 将通过 VPC 网络自动获得允许

对于连接详细信息，你将看到两个选项：公共网络和 VPC 网络。第一个用于像 MongoDB Compass 这样的工具进行外部访问，而第二个将由 Coreflux 服务用于访问数据库。

1. 记下提供的连接详细信息，包括公共访问和 VPC 访问（每种都有不同的详细信息）：

   • 主机：你的数据库主机名
   • 用户：默认管理员用户
   • 密码：自动生成的安全密码
   • 数据库：身份验证数据库名称

测试 MongoDB 数据库连接

你可以使用 MongoDB Compass 或提供的连接字符串，使用公共访问凭证测试 MongoDB 连接：

mongodb://username:password@mongodb-host:27017/defaultauthdb?ssl=true

创建 MongoDB 应用程序数据库和用户（可选）

为了更好的安全性和组织性，为你的 物联网自动化 应用程序创建一个专用用户和数据库。这也可以通过 MongoDB Compass 或 CLI 完成，但 DigitalOcean 提供了一种用户友好的方法：

1. 转到你的 托管数据库 集群中的 用户与数据库 选项卡

2. 创建用户：

   • 用户名：coreflux-broker-client
   • 密码：自动生成

3. 创建数据库：

   • 数据库名称：coreflux-broker-data

设置 OpenSearch 托管数据库

托管 OpenSearch 专为高容量物联网数据的搜索、日志分析和时间序列仪表板而设计，该服务为你管理集群健康、扩展和索引存储。

1. 从 DigitalOcean 控制面板，导航到 数据库
2. 点击 创建数据库集群

3. 为物联网自动化配置你的 OpenSearch 集群：

   • 数据库引擎：选择 OpenSearch
   • 版本：选择最新的稳定版本
   • 数据中心区域：选择法兰克福（与你的 VPC 相同）
   • VPC 网络：选择你创建的 coreflux-integrations-vpc
   • 数据库集群名称：opensearch-coreflux-test
   • 项目：选择你的目标项目

4. 根据你的 物联网 需求选择你的计划：

   1. 对于开发：基础 计划，1 GB RAM
   2. 对于生产：通用型 或更高，用于可扩展存储
5. 点击 创建数据库集群

托管数据库 创建过程通常需要 1-5 分钟。完成后，你将被重定向到数据库概览页面，在那里你可以查看连接详细信息并执行管理操作。

为 MQTT 代理集成配置 OpenSearch 数据库访问

系统将提示你进行入门步骤，显示你的连接详细信息，你可以配置入站访问规则（建议限制为你的 IP 和仅 VPC）。

1. 点击 开始使用 来配置你的 OpenSearch 数据库

2. （可选）限制入站连接：

   • 添加你本地计算机的 IP 以进行管理访问
   • droplet 将通过 VPC 网络自动获得允许

对于连接详细信息，你将看到两个选项：公共网络和 VPC 网络。第一个用于工具的外部访问，而第二个将由 Coreflux 服务用于访问数据库。你还将看到访问 OpenSearch 仪表板的 URL 和参数。

1. 记下提供的连接详细信息，包括公共访问和 VPC 访问（每种都有不同的详细信息）：

   • 主机：你的数据库主机名
   • 用户：默认管理员用户
   • 密码：自动生成的安全密码
   • 数据库：身份验证数据库名称

测试 OpenSearch 数据库连接

你可以使用提供的凭证通过 OpenSearch 仪表板测试 OpenSearch 连接：

步骤 3 — 在 DigitalOcean Droplet 上部署 Coreflux MQTT 代理

创建 DigitalOcean Droplet

1. 在你的 DigitalOcean 控制面板中导航到 Droplets
2. 点击 创建 Droplet

1. 为 MQTT 代理 部署配置你的 droplet：

   • 选择区域：法兰克福（与你的托管数据库相同）
   • VPC 网络：选择 coreflux-integrations-vpc
   • 选择镜像：转到 Marketplace 选项卡
   • 搜索 “Coreflux” 并从 Marketplace 中选择 Coreflux

1. 为你的 物联网 工作负载选择大小：

   • 对于开发：基础 计划，2 GB 内存
   • 对于生产：基础 或 通用型 计划，4+ GB 内存以获得可扩展性能

2. 选择身份验证方法：

   • SSH 密钥：推荐用于提高安全性

     1. 可以使用 ssh-keygen 在本地创建密钥
   • 密码：备选方案

3. 最终确定详细信息：

   • 主机名：coreflux-test-broker
   • 项目：选择你的项目
   • 标签：为 DevOps 组织添加相关标签
4. 点击 创建 Droplet
5. 查看 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 — 为安全的物联网通信配置防火墙规则（可选）

对于生产环境的物联网自动化部署，配置防火墙规则以限制访问：

1. 导航到网络 → 防火墙
2. 点击创建防火墙

3. 配置MQTT代理安全的入站规则：

   • SSH：来自你IP的端口22
   • MQTT：来自你的物联网应用程序源的端口1883
   • 带TLS的MQTT：用于安全的带TLS的MQTT的端口1884
   • WebSocket：用于通过WebSocket的MQTT的端口5000
   • 带TLS的WebSocket：用于通过带TLS的WebSocket的MQTT的端口443
4. 将防火墙应用到你的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)用于低代码物联网自动化的信息。

1. 打开Visual Studio Code
2. 转到扩展（Ctrl+Shift+X）
3. 搜索"LoT Notebooks"
4. 安装由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 — 验证完整的物联网自动化管道

监控实时数据流

1. MQTT Explorer：使用MQTT客户端验证实时数据发布
2. 数据库客户端：连接以验证数据的存储（PostgreSQL使用DBeaver，MongoDB使用MongoDB Compass，OpenSearch使用OpenSearch Dashboards）

验证PostgreSQL存储

使用DBeaver连接到你的PostgreSQL托管数据库以验证可扩展存储：

1. 使用来自你的DigitalOcean数据库的连接字符串
2. 导航到 coreflux-broker-data 数据库（或你为数据库指定的名称）
3. 检查 MachineProductionData 表中存储的记录

正如我们之前看到的，所有数据都可在MQTT代理中用于其他用途和集成。

验证MongoDB存储

使用MongoDB Compass连接到你的MongoDB托管数据库以验证可扩展存储：

1. 使用来自你的DigitalOcean数据库的连接字符串
2. 导航到 coreflux-broker-data 数据库（或你为数据库指定的名称）
3. 检查 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托管数据库以验证可扩展存储：

1. 使用来自你的DigitalOcean数据库的连接字符串
2. 导航到coreflux-broker-data数据库（或你为数据库指定的名称）
3. 检查MachineProductionData表中存储的记录

与其他集成一样，所有数据也可在MQTT代理中用于其他用途和下游集成。

验证OpenSearch存储

使用提供的URL和凭据打开OpenSearchDashboards：

1. 打开菜单并选择索引管理选项

   1. 在菜单中选择索引选项，查看你的表名是否出现在列表中

1. 返回主页并在菜单中选择发现选项

   1. 按照提供的步骤创建索引模式
   2. 返回到发现页面，你应该会看到你的数据

正如我们之前看到的，所有数据都可在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物联网数据的最佳托管数据库取决于你的数据结构、查询需求和分析目标。使用下面的比较表来帮助你决定：

提示： 你可以通过配置多个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之间做出选择？

你应该根据物联网数据结构、可扩展性以及你希望如何查询设备数据来选择托管数据库。下表总结了每个选项的优势：

结论

将Coreflux MQTT代理与DigitalOcean的托管数据库服务（PostgreSQL、MongoDB、MySQL或OpenSearch）集成，为你提供了实时物联网数据处理和存储的完整设置。按照本教程，你已经使用低代码开发实践构建了一个收集、处理和存储物联网数据的自动化管道。

借助Coreflux的架构和你选择的数据库的存储特性，你可以处理大量的实时数据并查询它以获取洞察。无论你是监控工业系统、跟踪环境传感器还是管理智慧城市基础设施，这种设置都让你能够基于实时MQTT主题和历史数据库查询做出数据驱动的决策。

了解更多关于DigitalOcean托管数据库的信息，以及DigitalOcean 针对 IoT行业的产品服务支持，可咨询 DigitalOcean 中国区独家战略合作伙伴卓普云AI Droplet（aidroplet.com）。

你可以尝试提供的用例或使用Coreflux和DigitalOcean实现你自己的用例。你还可以在DigitalOcean Droplet市场或通过Coreflux网站获取免费的Coreflux MQTT代理。

摘要：
OceanBase联合成都信息工程大学的数据库缺陷实证研究，被软件工程顶刊IEEE TKDE录用。研究聚焦复杂工作负载下数据库参数自动调优的效率与泛化能力问题，提出一种分层协作的多智能体框架，通过创新的训练机制与协同策略，有效破解传统调优方法的瓶颈，为数据库性能优化和预测提供了兼具理论创新性与工程实用性的技术方案。

近日，成都信息工程大学与 OceanBase 研发团队合作完成的研究《CMA+DB: How to Automatically Tune Database Parameters through Collaborative Multi-Agents》，被《IEEE Transactions on Knowledge and Data Engineering》（TKDE，CCF A 类、SCI 一区）录用。

研究聚焦复杂工作负载下数据库参数自动调优的效率与泛化能力问题，提出一种分层协作的多智能体框架，通过创新的训练机制与协同策略，有效破解传统调优方法的瓶颈，为数据库性能优化和预测提供了兼具理论创新性与工程实用性的技术方案。

以下为论文介绍：

研究背景与挑战

随着分布式与云计算技术的发展，数据库工作负载日趋多样化、复杂化。从高并发的在线交易场景（TPC-C），到随机读写的云服务场景（YCSB），再到高实时性的社交互动场景（Twitter），不同场景对参数配置的需求差异显著。

传统调优方式逐渐难以适配这些实际需求：人工调优高度依赖 DBA 的专业经验，不仅需要耗费大量时间梳理参数关系，而且无法应对动态变化的工作负载，往往出现 “调优即过时” 的问题；搜索式调优方法依赖启发式规则，在简单场景下表现尚可，但面对多参数交互的复杂场景时，搜索空间急剧扩大，调优效率大幅下降；贝叶斯优化方法需要手动筛选关键参数，若参数定义不全面，难以找到最优配置；现有强化学习调优方法多采用单智能体设计，仅能对参数进行粗粒度调优，无法充分挖掘不同类型参数间的深层交互影响，调优精度和泛化能力受限。

如何实现参数调优的自动化、精准化与高效化，成为数据库领域亟待解决的关键问题。

图1 数据库参数调优主要步骤

核心理论创新：CMA+DB 多智能体协作框架

CMA+DB 框架以 “分类协作、分层训练” 为核心设计理念，构建了三级递进式训练机制，整合单智能体预训练模型 SAPM、多智能体联合训练模型 MATM 与基于概率选择的联合训练模型 PJTM，既保障了单个参数类别内的调优深度，又实现了跨类别参数的协同优化。

三个子模型并非孤立存在，而是以级联方式递进工作，前一阶段的训练成果直接作为后一阶段的输入，形成 “基础专精—交互探索—精准强化” 的完整调优链路，确保框架在参数交互捕捉、调优效率与泛化能力之间实现最优平衡。

图2 CMA+DB 模型工作原理

在技术实现上，CMA+DB 基于深度确定性策略梯度（DDPG）与多智能体深度确定性策略梯度（MADDPG）构建核心算法架构，采用 Actor-Critic 网络结构实现智能体的决策与优化。每个智能体都具备独立的观测空间与动作空间，通过与数据库环境的持续交互获取反馈，不断优化参数调优策略。

图3 多智能体 Actor 和 Critic 网络结构

在第一阶段，提出单智能体预训练模型 SAPM。训练初期，每个 Agent 负责调优一类功能或参数级别相似的参数，目的是探究单个 Agent 对数据库性能的影响，进而优化其网络，使得其神经网络偏向于调节重要参数。该过程 Agent 之间相互不影响。与传统的单智能体模型相比，SAPM 模型的优点在于更深入地探究相似参数之间的关系，也更容易识别出一些重要参数。

这一阶段对于识别关键参数和实现快速收敛至关重要。通过使智能体能够初步掌握参数调整策略，SAPM 为后续更复杂的多智能体训练奠定了坚实的基础，并减少后续阶段的耦合干扰，从而提高了整体效率和效果。

在第二阶段，提出多智能体联合训练模型 MATM，在 MATM 模型训练阶段，使用组合算法将多个 Agent 的预测动作组合并映射为一组数据库的推荐配置，要求 Agent 之间不能存在相同的数据库参数。

这种协作方法不仅增加了可调参数的数量，还显著增强了模型的表现力和调整能力。此阶段虽然存在耦合，但通过联合训练，能够让智能体逐步适应彼此的行为模式，实现协同优化。

在第三阶段，提出了联合动作概选模型 PJTM，在 SAPM 和 MATM 模型训练之后进行训练，算法为每个 Agent 设置了一个概选因子 Pi， Agent 根据 Pi，随机地参与联合动作推荐。当一个 Agent 控制的数据库参数对数据库性能的提升微乎其微时，就可以根据 Pi 减小该 Agent 的动作参与，进而降低其对其它 Agent 所做动作的负面影响。这一机制有效减少了低影响智能体对整体调优过程的干扰，缓解了耦合带来的负面影响。

基于上述三种方式，提出协作型多智能体模型 CMA+DB，其整合 SAPM、MATM 和 PJTM 模型进行分阶段训练，实现分功能、分级别地对数据库参数进行调优。探究了相同类型参数之间的相互影响以及不同类型参数之间的协作关系。有效地解决了离散环境和异步反馈的挑战，确保多个智能体的动作是协调的，并针对数据库性能进行了优化，提高了数据库参数的调优数量，提升了数据库参数调优的效率。

关键验证成果

相关研究成果已在 PostgreSQL 数据库环境下，在 TPC-C、YCSB、Twitter 三种典型工作负载上，与主流调优方法进行全面对比验证，核心性能指标表现突出且稳定。

在调优效率方面，CMA+DB 的收敛速度优势显著。在 TPC-C 高并发交易场景中，这一优势尤为明显，CMA+DB 的平均收敛速度（达到最大吞吐量时）优于其他三种算法。能帮助数据库更快脱离性能波动期，进入最优稳定状态，大幅缩短调优耗时，降低系统上线或扩容后的性能适配成本。

在性能提升方面，CMA+DB 展现出持续优化的能力。经过 SAPM 预训练后，框架的吞吐量已优于现有部分主流方法；后续经 MATM 的跨类别参数协同优化，以及 PJTM 的精准强化，在 TPC-C 场景下，其延迟显著低于 OtterTune、CDBTune+ 等方法，有效保障了系统在峰值压力下的响应稳定性，避免了因参数配置不当导致的延迟突增问题。

在泛化能力方面，CMA+DB 表现出极强的适应性。实验结果表明，不同参数规模与智能体数量的组合测试显示，CMA+DB 能够根据实际场景灵活调整，无需人工干预即可适配不同规模的数据库参数调优需求，解决了传统方法在参数规模变化时泛化能力下降的问题。

总结与展望

作为一种兼具理论深度与工程价值的数据库参数自动调优方案，CMA+DB 框架通过多智能体分类协作与分层训练的创新设计，有效解决了传统方法在参数交互探索、调优效率与泛化能力上的短板，为 AI 赋能数据库优化领域提供了新的技术思路。

该框架的核心优势在于，不但实现了参数调优的自动化与精准化，而且无需依赖大量人工干预与领域知识，降低了数据库性能优化的门槛，具备较强的应用价值。

未来研究将进一步优化框架结构，通过算法改进降低计算复杂度，提升框架在超大规模参数场景下的运行效率；同时，将尝试适配 MySQL、OceanBase 等更多主流 DBMS，针对不同数据库的内核特性调整智能体的参数分类与协作策略，推动相关技术在金融、电商、社交等实际生产环境中的广泛应用，为数据库系统的性能优化提供更高效、更通用的解决方案。

欢迎访问 OceanBase 官网获取更多信息：https://www.oceanbase.com/

墨天轮社区举办的 【文档悬赏令】系列活动，每次聚焦一个主题，征集真实、可操作的第一手文档，希望能扩充社区文档资源、为更多人提供有用的参考资料，让技术学习少走弯路！

第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）重复文档：重新上传以前发布的文档系统会自动识别判定为重复仅自己可见，也不可删除旧文档后重新上传

三、上传步骤

1. 登录上传：登录墨天轮账号后，点击链接直达上传页：https://www.modb.pro/docUpload，或在首页下拉框点击 “传文档”；

2. 填写信息：

   • 标题：建议明确标注 “数据库种类 + 巡检场景”（如《MySQL数据库巡检手册》《Oracle数据库常规巡检项目和命令》），帮助他人快速判断主题
   • 标签：上传时需在 “标签” 栏填写 “数据库巡检” + 数据库种类（如 Oracle、OceanBase） 两个标签
   • 墨值设置：支持免墨值 / 5 墨值 / 10 墨值 / 25 墨值 / 50 墨值 / 100 墨值，设置墨值后，他人下载时支付的墨值将实时计入您的账户（ps：一般来说墨值越低下载门槛越低，被下载可能更高）

3. 提交：确认信息无误后点击 “提交文档”即可。于 “控制台-内容管理-我的文档” 处可查看您所有文档明细，若您遇到“仅自己可见”“转换失败”等状态可询问墨天轮小助手（VX：modb666）询问具体原因。

四、奖励设置

本次活动奖励分为 “合格奖”“有效助人奖”“巡检先锋奖” 三类

1、合格奖

每位用户每日首次上传≥5 页的合格文档，可自动触发 “每日任务” 获 5 墨值。此外，本次活动奖励额外可叠加，根据用户合格文档数量发放不同等级的墨值奖励，具体如下：

2、有效助人奖

根据单个文档下载数发放不同等级的墨值奖励，每个用户最多可有5个文档获得本项奖励：

3、巡检先锋奖

将综合评估用户上传的合格文档的内容质量及总下载量，评选出 “既多产、又优质” 的核心贡献者，发予相应奖励：

说明：三类奖项单独评奖、每位用户可重复获得。其中新版本先锋奖数量将根据实际参与情况灵活调整，若参与情况佳则可能增设、反之亦然。

五、奖励公布与发放

1. 进度公示：为了让大家更加了解自己的参与进度，将在活动期间不定期在活动原文评论区公布最新合格情况。若发现违规行为也欢迎向工作人员反馈，一经核实将取消其参与资格。
2. 结果公布：活动结束后 3 个工作日内公布所有获奖名单；
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号：数据库新版本的安装实操

欲了解更多可浏览墨天轮社区，围绕数据人的学习成长提供一站式的全面服务，打造集新闻资讯、在线问答、活动直播、在线课程、文档阅览、资源下载、知识分享及在线运维为一体的统一平台，持续促进数据领域的知识传播和技术创新。

关注官方公众号： 墨天轮、 墨天轮平台、墨天轮成长营、数据库国产化 、数据库资讯

本次版本 新增函数对象转换能力，扩展了达梦等多数据库迁移适配范围，并提升了批量转换的处理效率，进一步降低企业级数据库迁移的复杂度与成本。

一、核心特性

支持函数对象迁移

​函数对象可随存储过程的迁移任务一键同步转换​。该能力的加入，让 SQLShift^[1] 从一款“​存储过程迁移工具​”升级为“​核心业务逻辑对象全量迁移工具​”。随之也带来三重提升：

1. 降低迁移风险与人工成本

   避免 函数对象 需人工逐个改写与反复校验，大幅减少因语法差异、返回值不一致引发的运行期错误。

2. 提升非表对象整体迁移效率

   函数对象与存储过程 可在同一时间中完成迁移与校验，缩短整体迁移周期。

3. 保障业务逻辑完整性与可用性

避免 函数对象 缺失导致上层存储过程等对象无法编译或运行的问题，有效降低迁移后集中调试与返工压力，提升割接与上线的稳定性。

<iframe src="//player.bilibili.com/player.html?isOutside=true&aid=115970207123078&bvid=BV1Gg6LBAEf3&cid=35655845383&p=1" scrolling="no" border="0" frameborder="no" framespacing="0" allowfullscreen="true"></iframe>

新增数据库迁移组合

本次升级，SQLShift 扩展了多项数据库迁移组合。

• 新增 Oracle / OceanBase → 达梦

  降低了迁移至达梦数据库的复杂度和人工成本，帮助企业快速完成数据库替换或国产化改造。

• 新增 PostgreSQL → OceanBase（Oracle 模式）

  减少了跨数据库迁移中的人工调整工作量，加快了从 PostgreSQL 向 OceanBase 的迁移进程。

二、其他更新

批量处理能力提升

支持同时上传多个 SQL 文件进行转换，提升大规模迁移场景下的处理效率。

免费试用限时开放！

👉 点击领取 你的转换额度，立即体验 SQLShift 智能化迁移带来的飞跃效率！

🧩 SQL 方言再多，转换也能一步到位，SQLShift 为你搞定！

熟悉 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 查询。

1. 强大的查询优化：内置基于规则和成本的查询优化器，能自动优化 SQL 执行计划，提升查询效率，尤其是在复杂多表关联、跨数据源查询场景下，优化效果明显。
2. 灵活的数据源适配：通过“适配器（Adapter）”机制适配不同数据源，社区已提供大量现成适配器，也支持自定义开发，适配特殊数据源。

3. 轻量级集成：核心依赖体积小，无复杂依赖，可轻松集成到 Spring Boot、Spring Cloud 等主流 Java 开发框架中，无需单独部署独立服务（也支持独立部署）。

   二、重点实战：Spring Boot 3 集成 Calcite 核心步骤

   既然大家都熟悉 Spring Boot 3 的基础操作，我就不啰嗦项目搭建这些常规步骤了，直接聚焦 Calcite 集成的核心环节，每一步都附完整代码和避坑提醒，跟着做就能成！

2.1 核心依赖引入

第一步先引依赖，在 pom.xml 里加好 Calcite 核心包、对应数据源的适配器，再配上 MyBatis Plus 的核心依赖（替换掉原来的 Jdbc 依赖就行），具体如下：

<!-- Calcite 核心依赖 -->
<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-core</artifactId>
    <version>1.36.0</version> 
</dependency>

<!-- MySQL 适配器（用于适配 MySQL 数据源） -->
<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-mysql</artifactId>
    <version>1.36.0</version>
</dependency>

<!-- MongoDB 适配器（用于适配 MongoDB 数据源） -->
<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-mongodb</artifactId>
    <version>1.36.0</version>
</dependency>

<!-- Spring Boot 与 MyBatis Plus 集成核心依赖 -->
<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.5.5</version> <!-- 适配 Spring Boot 3 的稳定版 -->
</dependency>

<!-- 数据库连接池依赖（MyBatis Plus 需连接池支持） -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid-spring-boot-starter</artifactId>
    <version>1.2.20</version>
</dependency>

这里有 3 个避坑点必须强调下：

1. Calcite 所有组件版本要统一，不然容易出现类加载异常；
2. MyBatis Plus 得选适配 Spring Boot 3 的版本（3.5.3+）；

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"
      }
    }
  ]
}

几个关键配置给大家解释清楚，避免踩坑：

1. defaultSchema：默认查询的 Schema，可省略，查询时需指定 Schema 名称（如 ecommerce.order、user_mongo.user_info）。
2. factory：对应数据源的适配器工厂类，Calcite 已为主流数据源提供现成工厂，自定义数据源需实现自己的 Factory。

3. 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 个重点，确保大家少走弯路：

1. 实体类字段要和查询结果列名对应，用别名适配下划线转驼峰更省心；
2. Mapper 接口继承 BaseMapper 后，MyBatis Plus 的分页、条件构造器这些功能都能直接用，复杂查询也能轻松搞定；

3. 咱们写的都是标准 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

Kite：Kotlin/Java 通用的全自动 ORM 框架

Kite 是一个高效的轻量级 ORM 框架，基于 Kotlin 编写，开箱即用，内置分页查询、增删改查等常用功能，支持多表操作。它支持 PostgreSQL、MySQL、Derby 等多种数据库，旨在通过简化数据库操作，减少代码量，提升开发效率。

框架特点

• 全自动映射：无需手动编写 SQL，Kite 会自动根据实体类生成相应的数据库操作语句
• 支持自定义 SQL：在需要时，可以编写自定义 SQL 语句，满足复杂查询需求，还可以像写代码一样写流程控制语句
• 多数据库支持：支持 PostgreSQL、MySQL、Derby 等主流关系型数据库
• Kotlin/Java 双语言支持：既可以在 Kotlin 项目中使用，也可以在 Java 项目中无缝集成
• 轻量级设计：无过多依赖，性能优秀
• 丰富的 API：提供简洁直观的 API，支持各种复杂查询和操作
• Spring Boot 集成：提供 Spring Boot Starter，便于在 Spring Boot 项目中快速集成

使用方法（Spring Boot 集成示例）

Maven 中央仓库: kite-spring-boot-starter

1. 向项目添加以下依赖：

• Maven

<dependency>
   <groupId>io.github.tangllty</groupId>
   <artifactId>kite-spring-boot-starter</artifactId>
   <version>${kite.version}</version>
</dependency>

• Gradle

implementation("io.github.tangllty:kite-spring-boot-starter:${kite.version}")

2. 在数据库中创建表

使用 MySQL 演示

create table account (
  id          bigint not null auto_increment,
  username    varchar(32)     default '',
  password    varchar(32)     default '',
  balance     decimal(10,2)   default '0.00',
  create_time datetime        default null,
  update_time datetime        default null,
  primary key (`id`)
);

insert into account (username, password, create_time, balance) values
('admin', 'admin123', '2020-01-01 12:00:00', 1000.10),
('user', 'user123', '2024-05-02 8:30:00', 101.00),
('guest', 'guest123', '2022-03-03 15:00:00', 10.00),
('tang', 'tang123', '2019-06-01 21:30:30', 1.88),
('jeo', 'jeo123', '2024-07-01 5:59:59', 0.10);

3. 在 application.yml 文件中配置数据库连接信息

spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://127.0.0.1:3306/kite-test
    username: root
    password: password

4. 为 account 表创建模型类

• Java

import com.tang.kite.annotation.id.Id;
import com.tang.kite.annotation.id.IdType;
import java.math.BigDecimal;
import java.time.LocalDateTime;

public class Account {

    @Id(type = IdType.AUTO)
    private Long id;
    private String username;
    private String password;
    private BigDecimal balance;
    private LocalDateTime createTime;
    private LocalDateTime updateTime;

    // Getters and Setters
}

• Kotlin

import com.tang.kite.annotation.id.Id
import com.tang.kite.annotation.id.IdType
import java.math.BigDecimal
import java.time.LocalDateTime

class Account (

    @Id(type = IdType.AUTO)
    var id: Long? = null,
    var username: String? = null,
    var password: String? = null,
    var balance: BigDecimal? = null,
    var createTime: LocalDateTime? = null,
    var updateTime: LocalDateTime? = null

)

5. 继承 BaseMapper 接口创建 Mapper 接口

• Java

import com.tang.kite.mapper.BaseMapper;
import com.tang.kite.spring.annotation.Mapper;

@Mapper
public interface AccountMapper extends BaseMapper<Account> {
}

• Kotlin

import com.tang.kite.mapper.BaseMapper
import com.tang.kite.spring.annotation.Mapper

@Mapper
interface AccountMapper : BaseMapper<Account>

6. 在 Spring Boot 应用类上添加 @MapperScan 注解

• Java

import com.tang.kite.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@MapperScan("com.tang.application.mapper")
@SpringBootApplication
public class KiteApplication {

    public static void main(String[] args) {
        SpringApplication.run(KiteApplication.class, args);
    }

}

• Kotlin

import com.tang.kite.spring.annotation.MapperScan
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication

@MapperScan(["com.tang.application.mapper"])
@SpringBootApplication
class KiteApplication

fun main(args: Array<String>) {
    runApplication<KiteApplication>(*args)
}

7. 测试 Mapper 接口

• Java

import com.tang.demo.mapper.AccountMapper;
import com.tang.kite.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@MapperScan("com.tang.application.mapper")
@SpringBootApplication
public class KiteApplication {

    public static void main(String[] args) {
        var context = SpringApplication.run(KiteApplication.class, args);
        var accountMapper = context.getBean(AccountMapper.class);
        var accounts = accountMapper.select();
        accounts.forEach(System.out::println);
    }

}

• Kotlin

import com.tang.demo.mapper.AccountMapper
import com.tang.kite.spring.annotation.MapperScan
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication

@MapperScan(["com.tang.application.mapper"])
@SpringBootApplication
class KiteApplication

fun main(args: Array<String>) {
    val context = runApplication<KiteApplication>(*args)
    val accountMapper = context.getBean(AccountMapper::class.java)
    val accounts = accountMapper.select()
    accounts.forEach { println(it) }
}

文档与社区

官方文档

详细的使用文档请参考：

• 中文文档
• 英文文档

源码

Kite 的源码托管在 GitHub 和 Gitee 上，您可以在以下地址查看和贡献：

• Kite GitHub 仓库
• Kite Gitee 仓库

总结

Kite 是一个功能强大、易于使用的 ORM 框架，它通过全自动映射和简洁的 API，大大简化了数据库操作的开发工作。无论是在 Kotlin 项目还是 Java 项目中，都能提供高效、便捷的数据库访问体验。

如果您正在寻找一个轻量级、高性能的 ORM 框架，Kite 绝对值得一试！

关于作者：

Nickyoung，数据库领域从业者。PostgreSQL ACE，IvorySQL专家顾问委员会成员。

公众号 “ 👉 PostgreSQL 运维之道 ”。

给大家分享一个有趣的案例，同一个 sql，索引扫描比全表顺序扫描获取的数据更少。本篇我们深入分析一起索引排序规则损坏的案例，并 debug 验证索引扫描的主要过程。

问题现象

走索引扫描查询到 1 条数据。

testidx=# explain analyze select *  from user_info where userid ='1230005998';
                                                          QUERY PLAN                                                           
-------------------------------------------------------------------------------------------------------------------------------
 Index Scan using index_userid on user_info  (cost=0.28..35.61 rows=9 width=57) (actual time=0.030..0.032 rows=1 loops=1)
   Index Cond: ((userid)::text = '1230005998'::text)
 Planning Time: 0.118 ms
 Execution Time: 0.057 ms
(4 rows)

testidx=# select ctid,userid,region_id from user_info where userid ='1230005998';
  ctid  | userid    | region_id 
--------+----------------------+-----------
 (4,39) | 1230005998 | abc
(1 row)

不走索引顺序扫描查询到 11 条数据。

testidx=# set enable_indexscan to off;
SET
testidx=# explain analyze select * from user_info where userid ='1230005998';
                                                QUERY PLAN                                                
----------------------------------------------------------------------------------------------------------
 Seq Scanon user_info  (cost=0.00..51.50rows=9 width=57) (actual time=0.093..0.460rows=11 loops=1)
   Filter: ((userid)::text = '1230005998'::text)
   Rows Removed by Filter: 1309
 Planning Time: 0.116 ms
 Execution Time: 0.478 ms
(5rows)

testidx=# select ctid,userid,region_id from user_info where userid ='1230005998';
  ctid   | userid    | region_id 
---------+----------------------+-----------
 (4,39)  | 1230005998 | abc
 (9,14)  | 1230005998 | abc
 (9,32)  | 1230005998 | abc 
 (10,32) | 1230005998 | abc
 (12,5)  | 1230005998 | abc
 (26,23) | 1230005998 | abc
 (27,4)  | 1230005998 | abc
 (27,9)  | 1230005998 | abc
 (27,11) | 1230005998 | abc
 (34,38) | 1230005998 | abc
 (34,39) | 1230005998 | abc
(11rows)

testidx=#

对比两次查询结果，可以看到走索引扫描时，仅查询到第一条匹配的数据，对应 ctid 为(4,39)。索引损坏了？

问题分析

当我们怀疑索引损坏时，可以使用 amcheck 插件对索引进行扫描分析，检查是否存在异常。

可以看到 leaf page 8 的 itemoffset 24 和 25 违反了条目顺序不变性规则。即按照升序原则 24 号索引槽位对应的键值要小于等于 25 槽位，但经检查是大于的，所以排序规则混乱了。

testidx=# select * from bt_index_check('index_userid',true);
DEBUG:  StartTransaction(1) name: unnamed; blockState: DEFAULT; state: INPROGRESS, xid/subid/cid: 0/1/0
DEBUG:  verifying level 1 (true root level)
DEBUG:  verifying 7 items on internal block 3
DEBUG:  verifying level 0 (leaf level)
DEBUG:  verifying 207 items on leaf block 1
DEBUG:  verifying 204 items on leaf block 2
DEBUG:  verifying 204 items on leaf block 4
DEBUG:  verifying 204 items on leaf block 5
DEBUG:  verifying 204 items on leaf block 6
DEBUG:  verifying 235 items on leaf block 7
DEBUG:  verifying 78 items on leaf block 8
ERROR:  item order invariant violated for index "index_userid"
DETAIL:  Lower index tid=(8,24) (points to heap tid=(4,14)) higher index tid=(8,25) (points to heap tid=(9,14)) page lsn=1/331E9F98.
testidx=# 

使用 pageinspect 扩展，查看 leaf page 8 有 78 条记录，其中 itemoffset 24 和 25 对应的键值，24 的键值为'31 09 xxx'，25 的键值为'2b 4c xxx'，前者大，确实是有问题的。

testidx=# select * from bt_page_stats('index_userid',8);
 blkno | type | live_items | dead_items | avg_item_size | page_size | free_size | btpo_prev | btpo_next | btpo | btpo_flags 
-------+------+------------+------------+---------------+-----------+-----------+-----------+-----------+------+------------
     8 | l    |         78 |          0 |            31 |      8192 |      5356 |         7 |         0 |    0 |          1
(1 row)

testidx=# 

testidx=# select * from  bt_page_items('index_userid',8) where itemoffset in (22,23,24,25);
 itemoffset |  ctid  | itemlen | nulls | vars |                                  data                                   
------------+--------+---------+-------+------+-------------------------------------------------------------------------
         22 | (4,39) |      32 | f     | t    | 2b 4c 54 34 33 36 32 35 31 33 34 00 00 00 00 00 00 00 00 00 00 00 00 00
         23 | (4,9)  |      32 | f     | t    | 31 09 0d 0a 4c 54 34 33 36 32 35 31 33 34 37 33 36 30 30 37 39 30 30 32
         24 | (4,14) |      32 | f     | t    | 31 09 0d 0a 4c 54 34 33 36 32 35 31 33 34 37 33 36 30 30 37 39 30 30 32
         25 | (9,14) |      32 | f     | t    | 2b 4c 54 34 33 36 32 35 31 33 34 00 00 00 00 00 00 00 00 00 00 00 00 00
(4 rows)

testidx=#

明显的索引损坏了，怎么损坏的呢？

可能是 BUG 或者系统异常导致数据库 crash 等写坏， 还有一个glibc 版本差异导致索引损坏的场景，特别是 glibc 2.28 之前和之后的版本。

经过排查这次异常就是 glibc 差异导致的，glibc 版本从 2.17 到 2.28。

当遇到这样的索引损坏场景时，建议 reindex 对应的索引来修复。

这个问题基本分析清楚了，不过老杨不打算到此为止。 借此机会证实下索引扫描的逻辑，也搞清楚为什么仅扫描一条数据就结束。感兴趣的朋友可以继续往下看。

原理分析

btree 想必大家都很熟悉了（其实我很讨厌面试中对于 btree 的八股文，haha...）

再来回顾下结构，细节可以参考灿灿的书中btree 章节

检索的时候，从 root page 开始检索，在 leaf page 中找到键值匹配的 heap ctid，通过 ctid 去 heap 中 fetch 对应的数据。这里借用德哥画的图，来自github 博客。

另外 postgrespro 的博客btree 章节，对于检索过程描述的不错，推荐大家去看看。

例如查找等于 49 的数据，标黄部分及蓝色箭头描述了检索过程：从 root 节点出发，找到第一个匹配的 leaf 节点，顺着 leaf 节点的链表一直查找，直到检索完所有匹配的 leaf 节点。

简单回顾一些概念和原理后，我们上手 debug 来证实检索过程。

我们的检索条件为userid ='1230005998'

1. 先确定 first leaf page

btgettuple函数中首次扫描走_bt_first函数逻辑。

通常 leaf page 会有多个，扫描时通过二分查找，先找到键值匹配的目标 leaf page。 在\_bt_first 函数中，调用\_bt_search 函数，再调用\_bt_binsrch 函数进行二分查找。

初始的 low 为 1，high 为 8 对应 index_userid 这个索引的 leaf block 1 和 8

\_bt_compare 函数进行 key 匹配，这里 userid 为 text 类型，因此使用的比较函数为 bttextcmp

我们省去二分查找的过程，最终 high=low=8，确定目标数据在 leaf page 8

2、确定 first item

开始扫描目标 leaf page，同样采用二分查找，找到第一条匹配的 item。

\_bt_first 函数走到 offnum = \_bt_binsrch(rel, &inskey, buf)，在\_bt_binsrch 函数中初始 high 为 78，low 为 1（因为 leaf page 8 有 78 条 item）。

在多轮二分查找后，mid 为 22 时\_bt_compare 匹配到了预期数据。bttextcmp 函数中可以看到 text_cmp 入参 arg1, arg2 相同，都为 1230005998，result 为 0。

因此，low 为 22，high 为 22，找到了 first item。

3、遍历页面元组，设置扫描边界

while (offnum <= maxoff)循环，offnum 为 22，maxoff 为 78。

从 first item 即 offnum=22 开始遍历，\_bt_readpage 中调用\_bt_checkkeys 首次比较结果相同，itemIndex++为 1，continuescan 为 true，offnum 延顺到 Next 即 23。

循环中再次调用\_bt_checkkeys 进行比较，实际的比较函数为 texteq，offnum 为 23 时 key 值明显和检索条件的长度不同，值肯定是不同的，result 为 false。

result 传递给 test，因此*continuescan = false，\_bt_checkkeys 返回 false。

continuescan 为 false，因此 so->currPos.moreRight=false，so->currPos.firstItem = 0， so->currPos.lastItem = 1 - 1， so->currPos.itemIndex = 0;

就是这几个属性决定了扫描边界。 firstItem 和 lastItem 相同都为 0，说明扫描的范围就是 first Item 这一条数据。

index_getnext_slot 函数中根据 ctid(4,39)调用 index_fetch_heap 获取 heap 数据。

4、获取 next Item

btgettuple 函数中，后续扫描调用\_bt_next 函数。

so->currPos.moreRight 为 false，\_bt_readnextpage 函数 return false，因此\_bt_steppage 函数 return false

因此\_bt_next 函数返回 false，btgettuple 返回 false

index_getnext_tid 函数返回 NULL

tid 为 NULL，index_getnext_slot 函数返回 NULL

至此扫描结束。

从这个过程中可以看到，itemoffset 22 即记录 ctid(4,39)这条索引键值和检索条件匹配，但 23 不匹配，因此导致索引扫描结束，只扫描了一条数据。

从 seqscan 结果看，ctid (4,39)下一条符合条件的数据为(9,14)，对应到索引 itemoffset 25。从 bt_page_items 的结果来看，23 和 24 的键值是一样的，都比 25 大，因此索引排序规则是错乱的。

小结

本篇我们深入分析了一起索引排序规则损坏的案例，当出现类似问题时，可以利用 amcheck 和 pageinspect 扩展来分析解决。同时也 debug 证实了下索引扫描的一些关键过程。

HOW 2026 议题招募中

2026 年 4 月 27-28 日，由 IvorySQL 社区联合 PGEU（欧洲 PG 社区）、PGAsia（亚洲 PG 社区）共同打造的 HOW 2026（IvorySQL & PostgreSQL 技术峰会） 将再度落地济南。届时，PostgreSQL 联合创始人 Bruce Momjian 等顶级大师将亲临现场。

自开启征集以来，HOW 2026 筹备组已感受到来自全球 PostgreSQL 爱好者的澎湃热情。为了确保大会议题的深度与广度，我们诚邀您在 2026 年 2 月 27 日截止日期前，提交您的技术见解。

投递链接：https://jsj.top/f/uebqBc

我复现一下上面的操作说 zq-platform初始数据写不到数据库

第一次执行

alembic revision --autogenerate -m "init tables"

报错

INFO  [
            alembic.runtime.migration]
           Context impl PostgresqlImpl.INFO  [
            alembic.runtime.migration]
           Will assume transactional DDL.ERROR [
            alembic.util.messaging]
           Target database is not up to date.FAILED: Target database is not up to date.

意思是：你的数据库当前版本 (current) 落后于 Alembic 迁移脚本所定义的最新版本 (head) cnblogs.com+1。这就好比你手里拿着的是第3版的说明书，但产品已经更新到第5版了

要解决这个问题，核心思路就是将数据库的当前版本 (current) 更新到与最新的迁移脚本版本 (head) 一致。

1.查看当前数据库状态：首先，确认一下版本差异。在项目根目录下打开终端，依次运行：

    # 查看数据库当前记录的版本    alembic current    # 查看所有可用的迁移脚本版本（head）    alembic heads

你通常会看到 current 的版本号比 heads 的版本号要旧，或者 heads 显示了多个分支（这通常意味着存在多个分支迁移需要合并）。

分别显示

INFO  [
            alembic.runtime.migration]
           Context impl PostgresqlImpl.INFO  [
            alembic.runtime.migration]
           Will assume transactional DDL.

这证实了问题所在：数据库当前停留在一个空版本，并没有处于最新状态，所以 Alembic 拒绝你生成新的迁移脚本。

请直接运行下面这条命令来解决这个问题：

alembic upgrade head

这个命令会扫描 alembic/versions 文件夹，找到所有脚本，并依次在数据库中执行它们。

执行结果：

INFO  [
            alembic.runtime.migration]
           Context impl PostgresqlImpl.INFO  [
            alembic.runtime.migration]
           Will assume transactional DDL.INFO  [
            alembic.runtime.migration]
           Running upgrade  -> b6a31168d666, init tablesINFO  [
            alembic.runtime.migration]
           Running upgrade b6a31168d666 -> a79453452d83, add page design

数据库已经成功升级到最新版本了。从输出 Running upgrade b6a31168d666 -> a79453452d83 可以看到：数据库已经更新到了 a79453452d83，

打开数据库可以看到当前版本号：

重新生成迁移：现在可以再次尝试运行 alembic revision --autogenerate -m “init tables”。

alembic revision --autogenerate -m "init tables"

显示结果：

INFO  [
            alembic.runtime.migration]
           Context impl PostgresqlImpl.INFO  [
            alembic.runtime.migration]
           Will assume transactional DDL.Generating F:\下载程序与源码\★★★可执行项目收集★★★\zq-platform\backend-fastapi\alembic\versions\588
            bd64ec92e_init_tables.py
           ...  done

输出显示：Generating ... 588bd64ec92e\_init\_tables.py ... done

这表示 Alembic 成功扫描了你的代码和数据库，并发现它们之间的差异，自动生成了一个新的迁移脚本文件： 588bd64ec92e\_init\_tables.py。

接下来的步骤：让修改生效

现在脚本文件只是生成了，但数据库里的表结构还没变。你需要执行最后一步：

1. 查看生成的脚本（可选，但推荐）

你可以打开 alembic\versions\ 9abfcdbaff03\_init\_tables.py 看一眼。检查 upgrade() 函数里是否包含了 op.create\_table(…) 等语句。如果那里是空的，说明 Alembic 没检测到你的模型变化，或者模型没有被正确导入。

2. 更新数据库（必须执行）

在终端运行：

alembic upgrade head

这将会把刚才生成的脚本应用到你的 fastapi\_db 数据库中，创建对应的表结构。

显示

INFO  [
            alembic.runtime.migration]
           Context impl PostgresqlImpl.INFO  [
            alembic.runtime.migration]
           Will assume transactional DDL.INFO  [
            alembic.runtime.migration]
           Running upgrade a79453452d83 -> 588bd64ec92e, init tables

数据库迁移已经完全成功了！🎉

从输出 Running upgrade a79453452d83 -> 588bd64ec92e 可以确认：

你的 fastapi\_db 数据库已经成功应用了 init tables 脚本，所有定义的表结构现在都已经真实存在于数据库中了。

验证成果

如果你想确认表真的创建好了，可以：

1.打开 pgAdmin（如果你安装了的话）。

2.连接到 fastapi\_db。

3.展开 Schemas -> public -> Tables。

4.你应该能看到刚才定义的所有数据表（比如 users 等）。

或者，你可以直接在 Python 代码中试着向数据库写入一条数据，看是否报错。

再执行python scripts/ loaddata.py db\_init.json，导入数据，看到

导入完成:  成功: 38 条  失败: 0 条

先不要激动！！！日志最后一句“导入完成: 成功 38 条 / 失败 0 条”是脚本自己打印的统计，并不真实——

只要发生 ROLLBACK，整个事务就被回滚，数据库里一条新数据也没有写进去。

真正的失败原因就是这一条：

            asyncpg.exceptions.DataError:
            invalid input for query argument $4: '2026-01-11T19:44:39.752685'  (expected a 
            datetime.date
           or 
            datetime.datetime
           instance, got 'str')

也就是 core\_user.last\_login 字段传的是 字符串，而数据库列类型是 timestamp without time zone，异步驱动 asyncpg 不接受字符串隐式转换。

如何修复

def parse_datetime(value):    """解析日期时间字符串"""    if isinstance(value, str):        # 尝试多种日期时间格式        formats = [            "%Y-%m-%dT%H:%M:%S.%f",  # ISO 格式带微秒            "%Y-%m-%dT%H:%M:%S",      # ISO 格式不带微秒            "%Y-%m-%d %H:%M:%S.%f",   # 带微秒的空格分隔格式            "%Y-%m-%d %H:%M:%S",      # 不带微秒的空格分隔格式            "%Y-%m-%d",               # 仅日期格式        ]                for fmt in formats:            try:                return 
            datetime.strptime(value,
           fmt)            except ValueError:                continue                # 如果以上格式都不匹配，尝试 fromisoformat        try:            return 
            datetime.fromisoformat(value.replace(
          "Z", "+00:00"))        except ValueError:            pass                # 如果所有尝试都失败，返回原始值        return value    return value    ......    # 转换日期时间字段                for key, value in 
            fields.items():
                              if isinstance(value, str):                        # 检查是否为日期时间格式的字符串                        parsed_value = parse_datetime(value)                        # 如果成功解析且返回的是 datetime 对象，则替换原值                        if isinstance(parsed_value, datetime):                            fields[key] = parsed_value

再执行python scripts/ loaddata.py db\_init.json，直至这些数据都导入完成。

当看到

从文件导入数据: 
            db_init.json
          读取到 38 条记录2026-01-20 17:07:52,224 INFO 
            sqlalchemy.engine.Engine
           select 
            pg_catalog.version()
          2026-01-20 17:07:52,225 INFO 
            sqlalchemy.engine.Engine
           [raw sql] ()......2026-01-20 17:07:52,276 INFO 
            sqlalchemy.engine.Engine
           COMMIT导入完成:  成功: 38 条  失败: 0 条

·脚本成功读取了 db\_init.json 文件，识别出包含 38 条待导入的记录

·SQLAlchemy 引擎成功连接到 PostgreSQL 数据库（日志中出现 pg\_catalog.version() 是 PostgreSQL 特有的查询）

数据库验证

出现账号数据即为数据导入成功。

启动服务

python main.py或使用 uvicornuvicorn main:app --reload --host 0.0.0.0 --port 8000

这样初始数据写不到数据库问题就可以得到根本解决。

PostgreSQL 在各行各业的关键应用中具有极高适用性。尽管 PostgreSQL 提供了良好的性能，但仍存在一些用户不太关注但对整体效率与速度至关重要的问题。多数人认为增加 CPU 核数、更快的存储、更大内存即可提升性能，但还有同样重要的因素需要关注——那就是延迟。

延迟意味着什么？

数据库执行查询操作的耗时，仅占应用程序接收查询结果总耗时的极小部分。下图可直观呈现该过程的内在逻辑：

客户端应用发送请求后，驱动程序通过网络向 PostgreSQL 发送消息（a），数据库执行查询（b），并将结果集返回给应用程序（c）。关键问题在于：相较于查询执行时间（b），网络传输时间（a 与 c）是否具有显著影响。通过实验可以加以验证。

首先，使用 pgbench 初始化一个简单的测试数据库。对于本次测试，小规模数据库已足够：

cybertec$ pgbench -i blog
dropping old tables...
NOTICE:  table "pgbench_accounts" does not exist, skipping
NOTICE:  table "pgbench_branches" does not exist, skipping
NOTICE:  table "pgbench_history" does not exist, skipping
NOTICE:  table "pgbench_tellers" does not exist, skipping
creating tables...
generating data (client-side)...
vacuuming...
creating primary keys...
done in 0.19 s (drop tables 0.00 s, create tables 0.02 s, client-side generate 0.13 s, vacuum 0.02 s, primary keys 0.02 s).

随后进行第一次基础测试：建立单个 UNIX Socket 连接，运行 20 秒（只读测试）：

cybertec$ pgbench -c 1 -T 20 -S blog
pgbench (17.5)
starting vacuum...end.
transaction type: <builtin: select only>
scaling factor: 1
query mode: simple
number of clients: 1
number of threads: 1
maximum number of tries: 1
duration: 20 s
number of transactions actually processed: 1035095
number of failed transactions: 0 (0.000%)
latency average = 0.019 ms
initial connection time = 2.777 ms
tps = 51751.287839 (without initial connection time)

关键指标如下：

• 平均延迟：0.019 毫秒
• 每秒事务处理量（TPS）：51751

该数据表现对于单连接场景而言已属良好水平。

下一步执行相同查询测试，但将连接方式从 UNIX 套接字更换为指向本地主机（localhost）的 TCP 连接（非远程连接）：

cybertec$ pgbench -c 1 -T 20 -S blog -h localhost
pgbench (17.5)
starting vacuum...end.
transaction type: <builtin: select only>
scaling factor: 1
query mode: simple
number of clients: 1
number of threads: 1
maximum number of tries: 1
duration: 20 s
number of transactions actually processed: 583505
number of failed transactions: 0 (0.000%)
latency average = 0.034 ms
initial connection time = 3.290 ms
tps = 29173.916752 (without initial connection time)

结果出现明显变化，关键指标如下：

• 平均延迟：0.034 毫秒
• 每秒事务数（TPS）：29173

吞吐量下降约 44%。下图对此进行了直观展示：

值得注意的是，延迟仅从 0.019 毫秒上升至 0.034 毫秒，变化幅度极小。但由于查询本身执行速度极快，即便如此微小的延迟也会带来显著影响。执行计划可以说明这一点：

blog=# explain analyze SELECT *
      FROM   pgbench_accounts
WHERE  aid = 434232;
                         QUERY PLAN
------------------------------------------------------------
 Index Scan using pgbench_accounts_pkey on pgbench_accounts
   (cost=0.29..8.31 rows=1 width=97)
   (actual time=0.015..0.016 rows=0 l                                                                                                                  oops=1)
   Index Cond: (aid = 434232)
 Planning Time: 0.227 ms
 Execution Time: 0.047 ms
(4 rows)

执行计划中的关键数值为 0.016，表示索引扫描在表中定位记录所需的时间。将该数值与额外引入的网络延迟进行对比，即可理解微小变化为何会造成巨大差异。

真实网络环境中的延迟

在实际场景中，应用程序与数据库通常部署在不同的机器上。测试前，先查看 traceroute 的输出结果：

different_box$ traceroute 10.1.139.53
traceroute to 10.1.139.53 (10.1.139.53), 30 hops max, 60 byte packets
 1  _gateway (10.0.0.1)  0.212 ms  0.355 ms  0.378 ms
 2  cybertec (10.1.139.53)  0.630 ms  0.619 ms *

可以看到，从运行 pgbench 的主机到数据库服务器的路径较短，仅通过内部网络完成通信。

再次运行相同测试，结果如下：

different_box$ pgbench -h 10.1.139.53 -S -c 1 -T 20 blog
pgbench (17.5)
starting vacuum...end.
transaction type: <builtin: select only>
scaling factor: 1
query mode: simple
number of clients: 1
number of threads: 1
maximum number of tries: 1
duration: 20 s
number of transactions actually processed: 47540
number of failed transactions: 0 (0.000%)
latency average = 0.420 ms
initial connection time = 9.727 ms
tps = 2378.123901 (without initial connection time)

关键指标为：

• 平均延迟：0.420 毫秒
• 每秒事务数（TPS）：2378

即便延迟仅为 0.420 毫秒，吞吐量已从 5 万 TPS 降至 2378 TPS。虽然该测试仍为单连接，但原因十分清晰：网络传输所消耗的 0.4 毫秒，与索引读取所需的 0.016 毫秒相比，已是数量级上的差距。

下图展示了吞吐量变化情况：

可确定的是，若网络架构中增加更多网络层级，吞吐量数据将进一步显著下降。该问题在云计算环境中尤为突出，每一层负载均衡、每一次网络跳转、每一台路由设备、每一条防火墙规则，均会增加网络延迟，进而降低应用程序运行效率。对于执行耗时极短的查询操作而言，网络延迟产生的额外开销占比越高，查询操作本身的执行耗时占比则越低，其对整体性能的影响程度也随之下降。

并发机制：可行的解决方案？

上述实验展示了极端情况，适用于单一应用在应用与数据库间频繁交互的场景。而在负载较高的业务系统中，通常存在多用户并发访问的情况。若增加并发连接数，系统性能可呈现较为理想的表现：

cybertec$ pgbench -c 4 -j 4 -T 20 -S blog -h localhost
pgbench (17.5)
starting vacuum...end.
transaction type: <builtin: select only>
scaling factor: 1
query mode: simple
number of clients: 4
number of threads: 4
maximum number of tries: 1
duration: 20 s
number of transactions actually processed: 1639827
number of failed transactions: 0 (0.000%)
latency average = 0.049 ms
initial connection time = 5.637 ms
tps = 82007.653121 (without initial connection time)

提取关键数据如下：

• 平均延迟：0.429 毫秒
• 每秒事务数（TPS）：82007

使用 4 个并发连接，TPS 达到 82,000，增加更多并发可进一步提升。在现代服务器上，每秒超过 100 万次操作完全可行。但前提是数据库与查询来源距离接近，网络延迟不构成瓶颈。

更快的 CPU 是否有帮助？

常见疑问：增加 CPU 核数或提升单核性能是否有意义？对比如下：

• 索引查找：0.016 毫秒
• 网络延迟：0.490 毫秒

即便 CPU 更快，优化的仅为 0.016 毫秒，占总耗时约 3%，剩余 97% 时间不受影响。本质上，这与吞吐量关系不大，而是延迟问题。对于极短查询，延迟累积可能导致严重性能下降，尤其在云环境下网络复杂度更高。

对于执行时间较长的查询，延迟影响较小；但对于超快小查询，网络延迟可能成为主要性能瓶颈。

总结

延迟在高频、短时查询场景中具有决定性影响。单连接环境下，微小的网络延迟即可导致吞吐量大幅下降；通过并发可以在一定程度上缓解这一问题，但网络距离和拓扑结构仍是关键约束因素。相比之下，单纯提升 CPU 性能对以网络延迟为主导的场景改善有限。在云环境与分布式架构中，延迟问题需要在系统设计阶段予以重点关注。

原文链接：

https://www.cybertec-postgresql.com/en/postgresql-performance...

作者：Hans-Jürgen Schönig

HOW 2026 议题招募中

2026 年 4 月 27-28 日，由 IvorySQL 社区联合 PGEU（欧洲 PG 社区）、PGAsia（亚洲 PG 社区）共同打造的 HOW 2026（IvorySQL & PostgreSQL 技术峰会） 将再度落地济南。届时，PostgreSQL 联合创始人 Bruce Momjian 等顶级大师将亲临现场。

自开启征集以来，HOW 2026 筹备组已感受到来自全球 PostgreSQL 爱好者的澎湃热情。为了确保大会议题的深度与广度，我们诚邀您在 2026 年 2 月 27 日截止日期前，提交您的技术见解。

投递链接：https://jsj.top/f/uebqBc