> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-mintlify-fbfa8bee.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

> 该引擎支持将 ClickHouse 与 NATS 集成，以发布或订阅消息 subject，并在有新消息可用时进行处理。

# NATS 表引擎

该引擎支持将 ClickHouse 与 [NATS](https://nats.io/) 集成。

`NATS` 可让你：

* 发布或订阅消息 subject。
* 在有新消息可用时进行处理。

<div id="creating-a-table">
  ## 创建表
</div>

```sql theme={null}
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = NATS SETTINGS
    nats_url = 'host:port',
    nats_subjects = 'subject1,subject2,...',
    nats_format = 'data_format'[,]
    [nats_schema = '',]
    [nats_num_consumers = N,]
    [nats_queue_group = 'group_name',]
    [nats_secure = false,]
    [nats_max_reconnect = N,]
    [nats_reconnect_wait = N,]
    [nats_server_list = 'host1:port1,host2:port2,...',]
    [nats_skip_broken_messages = N,]
    [nats_max_block_size = N,]
    [nats_flush_interval_ms = N,]
    [nats_username = 'user',]
    [nats_password = 'password',]
    [nats_token = 'clickhouse',]
    [nats_credential_file = '/var/nats_credentials',]
    [nats_startup_connect_tries = 5,]
    [nats_max_rows_per_message = 1,]
    [nats_handle_error_mode = 'default']
```

必需参数：

* `nats_url` – host:port (例如 `localhost:4222`) 。
* `nats_subjects` – NATS 表要订阅/发布的 subject 列表。支持通配符 subject，例如 `foo.*.bar` 或 `baz.>`
* `nats_format` – 消息格式。使用与 SQL `FORMAT` 函数相同的表示法，例如 `JSONEachRow`。更多信息，请参见[Formats](/zh/reference/formats/index)部分。

可选参数：

* `nats_schema` – 如果格式需要 schema 定义，则必须使用此参数。例如，[Cap'n Proto](https://capnproto.org/) 需要提供 schema 文件的 path 以及根对象 `schema.capnp:Message` 的名称。
* `nats_stream` – NATS JetStream 中现有 stream 的名称。
* `nats_consumer_name` – NATS JetStream 中现有持久化拉取消费者的名称。
* `nats_num_consumers` – 每个表的消费者数量。默认值：`1`。仅适用于 NATS core：如果单个消费者的吞吐量不足，可指定更多消费者。
* `nats_queue_group` – NATS 订阅者的 queue group 名称。默认值为表名。
* `nats_max_reconnect` – 已弃用且不起作用；系统会按 `nats_reconnect_wait` timeout 永久执行重连。
* `nats_reconnect_wait` – 每次重连尝试之间的休眠时间 (毫秒) 。默认值：`2000`。
* `nats_server_list` - 用于 connection 的 server 列表。可用于连接到 NATS cluster。
* `nats_skip_broken_messages` - NATS 消息解析器对每个块中与 schema 不兼容消息的容忍数量。默认值：`0`。如果 `nats_skip_broken_messages = N`，则该引擎会跳过 *N* 条无法解析的 NATS 消息 (1 条消息等于 1 行数据) 。
* `nats_max_block_size` - 为从 NATS flush 数据而通过 poll 收集的行数。默认值：[max\_insert\_block\_size](/zh/reference/settings/session-settings#max_insert_block_size)。
* `nats_flush_interval_ms` - flush 从 NATS 读取的数据的 timeout。默认值：[stream\_flush\_interval\_ms](/zh/reference/settings/session-settings#stream_flush_interval_ms)。
* `nats_username` - NATS 用户名。
* `nats_password` - NATS 密码。
* `nats_token` - NATS 认证标记。
* `nats_credential_file` - NATS 凭据文件的 path。
* `nats_startup_connect_tries` - 启动时的连接尝试次数。默认值：`5`。
* `nats_max_rows_per_message` — 对于基于行的格式，一条 NATS 消息中写入的最大行数。默认值：`1`。
* `nats_handle_error_mode` — NATS 引擎的错误处理方式。可选值：default (如果消息解析失败，则抛出异常) ；stream (异常消息和原始消息将保存在虚拟列 `_error` 和 `_raw_message` 中) 。

SSL 连接：

如需使用安全连接，请设置 `nats_secure = 1`。
证书验证由 `CLICKHOUSE_NATS_TLS_SECURE` 环境变量控制；
如果证书已过期、为自签名证书、缺失或因其他原因无效，请将 `CLICKHOUSE_NATS_TLS_SECURE=0` 以禁用验证。

写入 NATS 表：

如果表仅读取一个 subject，任何 insert 都会发布到同一个 subject。
但是，如果表从多个 subject 读取，则需要指定要发布到哪个 subject。
因此，向具有多个 subject 的表执行 insert 时，需要设置 `stream_like_engine_insert_queue`。
你可以选择该表读取的其中一个 subject，并将数据发布到该 subject。例如：

```sql theme={null}
  CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1,subject2',
             nats_format = 'JSONEachRow';

  INSERT INTO queue
  SETTINGS stream_like_engine_insert_queue = 'subject2'
  VALUES (1, 1);
```

也可以在添加与 nats 相关的设置时，一并添加 format 设置。

示例：

```sql theme={null}
  CREATE TABLE queue (
    key UInt64,
    value UInt64,
    date DateTime
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1',
             nats_format = 'JSONEachRow',
             date_time_input_format = 'best_effort';
```

可以通过 ClickHouse 配置文件添加 NATS 服务器配置。
更具体地说，你可以为 NATS 引擎添加密码：

```xml theme={null}
<nats>
    <user>click</user>
    <password>house</password>
    <token>clickhouse</token>
</nats>
```

<div id="description">
  ## 说明
</div>

`SELECT` 并不特别适合用于读取消息 (调试除外) ，因为每条消息只能读取一次。更实用的做法是使用 [materialized views](/zh/reference/statements/create/view) 创建实时处理链路。为此：

1. 使用该引擎创建一个 NATS 消费者，并将其视为数据 stream。
2. 创建一个具有所需结构的表。
3. 创建一个 materialized view，将该引擎中的数据转换后写入前面创建的表中。

当 `MATERIALIZED VIEW` 连接到该引擎后，就会开始在后台收集数据。这样一来，你就可以持续接收来自 NATS 的消息，并使用 `SELECT` 将其转换为所需格式。
一个 NATS 表可以拥有任意数量的 materialized view；它们不会直接从该表读取数据，而是接收新的记录 (以块的形式) ，因此你可以写入多个明细粒度不同的表 (带分组聚合和不带分组聚合) 。

示例：

```sql theme={null}
  CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1',
             nats_format = 'JSONEachRow',
             date_time_input_format = 'best_effort';

  CREATE TABLE daily (key UInt64, value UInt64)
    ENGINE = MergeTree() ORDER BY key;

  CREATE MATERIALIZED VIEW consumer TO daily
    AS SELECT key, value FROM queue;

  SELECT key, value FROM daily ORDER BY key;
```

若要停止接收流数据或更改转换逻辑，请分离 materialized view：

```sql theme={null}
  DETACH TABLE consumer;
  ATTACH TABLE consumer;
```

如果你想使用 `ALTER` 修改目标表，我们建议先禁用物化视图，以避免目标表与视图数据之间出现不一致。

<div id="virtual-columns">
  ## 虚拟列
</div>

* `_subject` - NATS 消息的 subject。数据类型：`String`。

当 `nats_handle_error_mode='stream'` 时，会提供以下额外的虚拟列：

* `_raw_message` - 无法成功解析的原始消息。数据类型：`Nullable(String)`。
* `_error` - 解析失败时产生的异常消息。数据类型：`Nullable(String)`。

注意：`_raw_message` 和 `_error` 这两个虚拟列仅会在解析过程中发生异常时填充；如果消息解析成功，它们始终为 `NULL`。

<div id="data-formats-support">
  ## 数据格式支持
</div>

NATS 引擎 支持 ClickHouse 支持的所有[格式](/zh/reference/formats/index)。
单条 NATS 消息中的行数取决于格式是按行还是按块：

* 对于按行的格式，可通过设置 `nats_max_rows_per_message` 来控制单条 NATS 消息中的行数。
* 对于按块的格式，无法将块拆分成更小的部分，但一个块中的行数可通过通用设置 [max\_block\_size](/zh/reference/settings/session-settings#max_block_size) 控制。

<div id="using-jetstream">
  ## 使用 JetStream
</div>

在结合 NATS JetStream 使用 NATS 引擎之前，您必须先创建一个 NATS stream 和一个持久化拉取消费者。为此，您可以使用 [NATS CLI](https://github.com/nats-io/natscli) 包中的 `nats` 工具，例如：

<Accordion title="创建 stream">
  ```bash theme={null}
  $ nats stream add
  ? Stream Name stream_name
  ? Subjects stream_subject
  ? Storage file
  ? Replication 1
  ? Retention Policy Limits
  ? Discard Policy Old
  ? Stream Messages Limit -1
  ? Per Subject Messages Limit -1
  ? Total Stream Size -1
  ? Message TTL -1
  ? Max Message Size -1
  ? Duplicate tracking time window 2m0s
  ? Allow message Roll-ups No
  ? Allow message deletion Yes
  ? Allow purging subjects or the entire stream Yes
  Stream stream_name was created

  Information for Stream stream_name created 2025-10-03 14:12:51

                  Subjects: stream_subject
                  Replicas: 1
                   Storage: File

  Options:

                 Retention: Limits
           Acknowledgments: true
            Discard Policy: Old
          Duplicate Window: 2m0s
                Direct Get: true
         Allows Msg Delete: true
              Allows Purge: true
    Allows Per-Message TTL: false
            Allows Rollups: false

  Limits:

          Maximum Messages: unlimited
       Maximum Per Subject: unlimited
             Maximum Bytes: unlimited
               Maximum Age: unlimited
      Maximum Message Size: unlimited
         Maximum Consumers: unlimited

  State:

                  Messages: 0
                     Bytes: 0 B
            First Sequence: 0
             Last Sequence: 0
          Active Consumers: 0
  ```
</Accordion>

<Accordion title="创建持久化拉取消费者">
  ```bash theme={null}
  $ nats consumer add
  ? Select a Stream stream_name
  ? Consumer name consumer_name
  ? Delivery target (empty for Pull Consumers) 
  ? Start policy (all, new, last, subject, 1h, msg sequence) all
  ? Acknowledgment policy explicit
  ? Replay policy instant
  ? Filter Stream by subjects (blank for all) 
  ? Maximum Allowed Deliveries -1
  ? Maximum Acknowledgments Pending 0
  ? Deliver headers only without bodies No
  ? Add a Retry Backoff Policy No
  Information for Consumer stream_name > consumer_name created 2025-10-03T14:13:51+03:00

  Configuration:

                      Name: consumer_name
                 Pull Mode: true
            Deliver Policy: All
                Ack Policy: Explicit
                  Ack Wait: 30.00s
             Replay Policy: Instant
           Max Ack Pending: 1,000
         Max Waiting Pulls: 512

  State:

    Last Delivered Message: Consumer sequence: 0 Stream sequence: 0
      Acknowledgment Floor: Consumer sequence: 0 Stream sequence: 0
          Outstanding Acks: 0 out of maximum 1,000
      Redelivered Messages: 0
      Unprocessed Messages: 0
             Waiting Pulls: 0 of maximum 512
  ```
</Accordion>

创建好 stream 和持久化拉取消费者后，我们就可以创建一个使用 NATS 引擎的表。为此，您需要设置：nats\_stream、nats\_consumer\_name 和 nats\_subjects：

```SQL theme={null}
CREATE TABLE nats_jet_stream (
    key UInt64,
    value UInt64
  ) ENGINE NATS 
    SETTINGS  nats_url = 'localhost:4222',
              nats_stream = 'stream_name',
              nats_consumer_name = 'consumer_name',
              nats_subjects = 'stream_subject',
              nats_format = 'JSONEachRow';
```