# 将 MQTT 数据写入 Disk Log

Disk Log 数据集成功能允许 EMQX 以 [JSON Lines](https://jsonlines.org/) 格式将事件数据持久化写入本地磁盘，类似于传统的轮转日志（rotating log）机制。这种方式支持长期事件留存，便于故障排查和历史数据分析。

本文将详细介绍 EMQX 与 Disk Log 的数据集成方式，并提供规则与 Sink 的配置指导。

## 工作原理

EMQX 提供内置的系统日志功能，用于记录错误、警告等运行时事件，而 Disk Log 集成则面向不同的目标：它用于将 MQTT 消息数据和客户端事件实际写入磁盘，以实现存储和离线处理。

Disk Log 集成基于 EMQX 的[规则引擎](./rules.md)和 Sink 机制实现，用户可以灵活定义捕获哪些数据，以及如何进行存储：

1. 使用规则对 MQTT 消息或客户端事件进行过滤、转换和提取。
2. 将 Disk Log Sink 添加到规则，指定数据的存储路径与格式，Sink 会将格式化后的 JSON 数据传递给连接器。
3. Disk Log 连接器负责将数据实际写入文件系统，并控制日志路径、文件轮转策略等配置。
4. 当规则被触发并产生匹配数据后，Sink 会调用所配置的连接器，将数据以 JSON Lines 格式写入根据所配置的基础文件路径生成的轮转文件。每条消息以一行 JSON 对象存储，便于标准工具或下游系统处理。

### 日志轮转机制

Disk Log 集成将消息写入本地文件系统中的一组轮转文件。配置的**日志文件路径**是基础文件路径，而不是目录。例如，基础路径为 `/var/log/emqx/mqtt-trace.log` 时，实际日志文件带有数字后缀，如 `mqtt-trace.log.1`。

Disk Log 支持按大小轮转和按时间轮转。按大小轮转始终生效：

- **文件大小限制**：当当前日志文件达到配置的最大大小后，EMQX 会创建新文件继续写入。
- **文件数量限制**：当日志文件数量达到上限后，EMQX 会丢弃最旧文件的内容，并将该文件用于写入新记录。
- **数据完整性保证**：每个日志文件至少包含一条完整的消息记录，即使该条记录超过了文件大小限制。

按时间轮转会在每个小时或每天的边界创建一组独立的文件。可通过以下连接器字段配置：

| Dashboard 字段 | 配置项 | 描述 | 默认值 |
| --- | --- | --- | --- |
| **轮转周期** | `rotation.period` | 选择 `hour` 或 `day`，在每个周期边界创建一组文件。选择 `None`（`none`）时仅按大小轮转。 | `None`（`none`） |
| **保留时间** | `rotation.retention_period` | 设置 EMQX 保留之前周期文件的时长。连接器启动和每次周期轮转后，EMQX 会删除超过保留期的文件。`rotation.period` 为 `none` 时，此参数不生效。 | `infinity` |
| **轮转时区** | `rotation.timezone` | 设置用于确定周期边界和文件名时间戳的时区。支持 `UTC`、`local` 或固定 UTC 偏移量，如 `+02:00`。 | `UTC` |

启用按时间轮转后，EMQX 会在文件扩展名前插入连字符（`-`）和 `YYYYMMDDHH` 格式的时间戳。按小时轮转时，文件名包含对应小时，例如 `mqtt-trace-2026062413.log.1`。按天轮转时，小时部分固定为 `00`，例如 `mqtt-trace-2026062400.log.1`。文件名始终带有 `.N` 后缀。每个周期的文件集还包含 `.idx` 和 `.siz` 管理文件。

**最大文件大小**和**最大文件数量**分别应用于每个周期。为避免同一周期内的文件被覆盖，应根据每小时或每天的最大预期数据量配置足够的容量。每个周期的近似容量为**最大文件大小**乘以**最大文件数量**。

EMQX 在连接器健康检查时检测周期边界。因此，新周期文件会在周期边界后的一个 `resource_opts.health_check_interval` 内生成。

以下示例在 `base.hocon` 中配置按天轮转，并将文件保留 30 天：

```hocon
connectors.disk_log.my_disk_log {
  filepath = "/var/log/emqx/mqtt-trace.log"
  max_file_size = "1GB"
  max_file_number = 24
  rotation {
    period = day
    retention_period = "30d"
    timezone = "UTC"
  }
}
```

::: warning 重要提示

不要对正在使用的 Disk Log 文件应用带有 `copytruncate` 的外部 `logrotate` 规则。外部截断不会重置 `.siz` 文件记录的大小，可能导致非预期轮转。如需按小时或按天生成文件，请使用内置的按时间轮转功能。

:::

## 特性与优势

Disk Log 提供一种轻量、本地优先的 MQTT 消息持久化方式，适用于多种场景。其主要优势包括：

- **精细化数据控制**：通过 SQL 规则捕获所需的消息或事件；支持在写入前对数据进行过滤、转换与丰富。
- **结构化输出格式**：采用 JSON Lines 格式存储，每行一条 JSON 记录，便于使用命令行工具或导入分析系统。
- **轻量独立部署**：无需依赖外部数据库或云存储。
- **增强可观测性与调试能力**：支持消息级别追踪，适用于审计、测试与故障排查。它记录的是业务数据，作为系统日志的有力补充。

## 准备工作

在配置 Disk Log Sink 之前，请完成以下准备工作：

### 前置条件

- 了解[规则引擎](./rules.md)
- 熟悉[数据集成](./data-bridges.md) 概念

### 创建日志目录

在部署 EMQX 的主机上创建一个用于存储日志文件的目录。确保运行 EMQX 的系统用户对该目录具有读写权限。

## 创建连接器

在创建 Disk Log Sink 之前，需先创建对应的连接器。

1. 在 Dashboard 左侧导航栏中，点击**集成** -> **连接器**。
2. 点击右上角的**创建**。
3. 选择 **Disk Log** 作为连接器类型，点击 **下一步**。
4. 输入连接器名称（如 `my-disk-log`），支持字母和数字组合。
5. 填写连接器参数：
   - **日志文件路径**：日志文件的基础路径，例如 `/var/log/emqx/mqtt-trace.log`。运行 EMQX 的系统用户必须对文件所在目录具有写入权限。
   - **最大文件大小**：单个日志文件的最大大小，超过后触发轮转。注意：每个日志文件至少会写入一条记录，因此如果单条日志记录超过此值，最终文件大小可能会超过此最大值。
   - **最大文件数量**：每组文件最多包含的文件数。启用按时间轮转后，此限制分别应用于每个小时或每天。
   - **轮转周期**：选择 `hour` 或 `day`，在每个周期边界创建一组文件。选择 `None` 时仅按大小轮转。
   - **保留时间**：设置之前周期文件的保留时长。默认值为 `infinity`。
   - **轮转时区**：设置用于确定周期边界和文件名时间戳的时区。支持 `UTC`、`local` 或固定 UTC 偏移量，如 `+02:00`。轮转配置详情参见[日志轮转机制](#日志轮转机制)。
6. 可点击**测试连接**验证 EMQX 是否具备向该目录写入的权限。
7. 点击**创建**完成连接器配置。

## 创建 Disk Log Sink 规则

以下步骤展示如何创建一条规则，将来自主题 `t/#` 的 MQTT 消息写入本地日志文件。

1. 进入 Dashboard 的**集成** -> **规则**页面。

2. 点击右上角的**创建**。

3. 设置规则 ID 为 `my_rule`，并在 SQL 编辑器中输入以下规则语句：

   ```sql
   SELECT
     *
   FROM
     "t/#"
   ```

   ::: tip

   如不熟悉 SQL，可点击 **SQL 示例**或**启用调试**测试输出结果。

   :::

4. 添加动作，选择 **Disk Log** 作为动作类型，可选择创建新 Sink 或使用已有 Sink。此处以创建新 Sink 为例。

5. 输入 Sink 名称和描述。

6. 从下拉列表中选择之前创建的连接器 `my-disk-log`，也可点击旁边的按钮快速新建。

7. 选择**写入模式**（`异步`或`同步`）。

8. 配置**消息模板**，内容必须能渲染为合法 JSON 对象。

9. **备选动作（可选）**：如果您希望在消息投递失败时提升系统的可靠性，可以为 Sink 配置一个或多个备选动作。当 Sink 无法成功处理消息时，这些备选动作将被触发。更多信息请参见：[备选动作](./data-bridges.md#备选动作)。

10. 如需高级配置，可展开[**高级设置**](#高级设置)选项。

11. 保持其他配置默认，点击**创建**完成 Sink 配置。页面将返回规则配置，新的 Sink 会自动添加为动作。

12. 最后点击**创建**，完成整条规则的创建。

规则创建完成后，您可以在**规则**页面看到新建规则，并在 **动作（Sink）** 标签页中查看对应的 Sink。

此外，还可前往**集成** -> **Flow 设计器**，可视化查看从主题 `t/#` 到磁盘写入的完整数据流路径。

## 测试规则

以下示例演示如何使用 MQTTX 发布测试消息，并验证是否成功写入日志文件。

执行以下命令向主题 `t/1` 发送测试消息：

```bash
mqttx pub -i emqx_c -t t/1 -m '{ "msg": "Hello Disk Log" }'
```

多次发送消息后，检查配置的日志目录中最新修改的文件，确认其中包含所发布的消息内容。

## 高级设置

在创建或编辑 Disk Log Sink 时，您可以展开**高级设置**，配置如下可调参数以满足更高性能或可靠性要求：

| 参数名称             | 描述                                                         |
| -------------------- | ------------------------------------------------------------ |
| **缓存池大小**       | 设置用于处理写入请求的工作进程数量，提升高并发场景下的吞吐性能。 默认值：`16` |
| **请求超期**         | 缓存中的请求在多长时间内有效（单位：秒）。若超时未被写入则视为失败。 |
| **健康检查间隔**     | Sink 自动对磁盘写入状态进行健康检查的周期（单位：秒）。 默认值：`15` |
| **缓冲队列最大长度** | 每个工作进程可缓冲的最大数据量（字节）。根据消息大小和性能调优。 默认值：`256` |
| **请求模式**         | 写入方式，可选`同步`或`异步`。异步模式下不会阻塞 MQTT 发布，但存在消息未落盘即确认的风险。 默认值：`异步` |
| **最大批量请求大小** | 每次批量写入的最大消息数量。值越大吞吐性能越好。设置为 `1` 则逐条写入。 默认值：`1000` |
| **请求飞行窗口**     | 异步请求最大并发数，控制 Sink 的并发写入能力。如需保障消息顺序，应设为 `1`。 默认值：`100` |