[comment]: # ({cdb0d4ea-b45624e8}) # 2 将数据流传输到外部系统 [comment]: # ({/cdb0d4ea-b45624e8}) [comment]: # ({fa81590c-c479d2fd}) #### 概述 可以通过 HTTP 将监控项值和事件从 Zabbix 流式传输到外部系统（请参见 [协议详情](#protocol)）。 标签过滤器可用于流式传输监控项值或事件的子集。 有两种 Zabbix 服务器进程类型负责数据流式传输：`connector manager` 和 `connector worker`。 Zabbix 内部监控项 `zabbix[connector_queue]` 允许监控已入队到 connector 队列中的值数量。 [comment]: # ({/fa81590c-c479d2fd}) [comment]: # ({695ba0a3-cfad3c2f}) #### 配置 配置向外部系统进行数据流转需要以下步骤： 1\. 准备一个用于接收来自 Zabbix 数据的远程系统。 为此，可使用以下工具： - 一个简单的 [receiver](https://git.zabbix.com/projects/ZT/repos/receiver/browse) 示例，它会将接收到的信息记录到 `events.ndjson` 和 `history.ndjson` 文件中。 - [Zabbix 服务器的 Kafka connector](https://git.zabbix.com/projects/ZT/repos/kafka-connector/browse) - 一个用 Go 编写的轻量级服务器，旨在将来自 Zabbix 服务器的监控项值和事件转发到 Kafka broker。 2\. 通过调整 `zabbix_server.conf` 中的 [`StartConnectors`](/manual/appendix/config/zabbix_server#startconnectors) 参数，在 Zabbix 中设置所需数量的 connector worker。 connector worker 的数量应与 Zabbix 前端中配置的 connector 数量一致（如果并发会话数大于 1，则应更大）。 然后，重启 Zabbix 服务器。 3\. 在 Zabbix 前端（*Administration* > *General* > *Connectors*）中配置一个新的 connector，并使用 `zabbix_server -R config_cache_reload` 命令重新加载服务器缓存。 ![](../../../../assets/en/manual/config/connector.png){width="600"} 必填字段以星号标记。 |Parameter|Description| |--|--------| |*Name*|输入 connector 名称。| |*Data type*|选择要流转的数据类型：
**Item values** - 将 Zabbix 中的监控项值流转到外部系统；
**Events** - 将 Zabbix 中的事件流转到外部系统。| |*URL*|输入 receiver URL。支持用户宏。| |*Tag filter*|仅导出符合标签过滤条件的监控项值或事件。如果未设置，则导出全部内容。
可以包含或排除特定标签及标签值。可设置多个条件。标签名称匹配始终区分大小写。

每个条件可使用以下运算符：
**Exists** - 包含指定的标签名称；
**Equals** - 包含指定的标签名称和值（区分大小写）；
**Contains** - 包含标签值中含有输入字符串的指定标签名称（子串匹配，不区分大小写）；
**Does not exist** - 排除指定的标签名称；
**Does not equal** - 排除指定的标签名称和值（区分大小写）；
**Does not contain** - 排除标签值中含有输入字符串的指定标签名称（子串匹配，不区分大小写）。

条件有两种计算方式：
**And/Or** - 必须满足所有条件，具有相同标签名称的条件将按 Or 条件分组；
**Or** - 满足任一条件即可。| |*Type of information*|选择信息类型（无符号数值、浮点数值、字符等），用于过滤 connector 应流转的监控项值。
当 *Data type* 设置为 "Item values" 时，此字段可用。| |*HTTP authentication*|选择认证方式：
**None** - 不使用认证；
**Basic** - 使用基本认证；
**NTLM** - 使用 NTLM ([Windows NT LAN Manager](http://en.wikipedia.org/wiki/NTLM)) 认证；
**Kerberos** - 使用 Kerberos 认证（另请参见：[使用 Zabbix 配置 Kerberos](/manual/appendix/items/kerberos)）；
**Digest** - 使用 Digest 认证；
**Bearer** - 使用 Bearer 认证。| |*Username*|输入用户名（最多 255 个字符）。支持用户宏。
当 *HTTP authentication* 设置为 "Basic"、"NTLM"、"Kerberos" 或 "Digest" 时，此字段可用。| |*Password*|输入用户密码（最多 255 个字符）。支持用户宏。
当 *HTTP authentication* 设置为 "Basic"、"NTLM"、"Kerberos" 或 "Digest" 时，此字段可用。| |*Bearer token*|输入 Bearer token。支持用户宏。
当 *HTTP authentication* 设置为 "Bearer" 时，此字段可用且必填。| |*Advanced configuration*|单击 *Advanced configuration* 标题以显示高级配置选项（见下文）。| |*Max records per message*|指定单条消息中可流转的最大值或事件数量。| |*Concurrent sessions*|选择此 connector 要运行的发送进程数量。最多可指定 100 个会话；默认值为 "1"。| |*Attempts*|流转数据的尝试次数。最多可指定 5 次尝试；默认值为 "1"。| |*Attempt interval*|指定 connector 在一次流转数据尝试失败后应等待的时间。最多可指定 10s；默认值为 "5s"。
当 *Attempts* 设置为 "2" 或更大时，此字段可用。
尝试失败是指建立连接失败，或 HTTP 响应码不是 200、201、202、203、204。若发生通信错误，或 HTTP 响应码不是 200、201、202、203、204、400、401、403、404、405、415、422，则会**触发**重试。重定向会被跟随，因此 302 -> 200 视为成功响应；而 302 -> 503 将触发重试。| |*Timeout*|指定消息超时时间（1-60 秒，默认值为 5 秒）。
支持时间后缀，例如 30s、1m。支持用户宏。| |*HTTP proxy*|可按以下格式指定要使用的 HTTP proxy：
`[protocol://][username[:password]@]proxy.example.com[:port]`
支持用户宏。

可选的 `protocol://` 前缀可用于指定其他 proxy 协议（该协议前缀支持已在 cURL 7.21.7 中加入）。如果未指定协议，则该 proxy 将被视为 HTTP proxy。默认使用 1080 端口。

如果指定了 *HTTP proxy*，该 proxy 将覆盖 `http_proxy`、`HTTPS_PROXY` 等与 proxy 相关的环境变量。如果未指定，则不会覆盖与 proxy 相关的环境变量。输入的值会原样传递，不会进行有效性检查。
也可以输入 SOCKS proxy 地址。如果指定了错误的协议，connector 将无法从 Zabbix 流转监控项值或事件。

请注意，HTTP proxy 仅支持简单认证。| |*SSL verify peer*|勾选此复选框以验证 web 服务器的 SSL 证书。
服务器证书将自动从系统范围的证书颁发机构（CA）位置获取。可使用 Zabbix 服务器或 proxy 配置参数 [`SSLCALocation`](/manual/appendix/config/zabbix_server#sslcalocation) 覆盖 CA 文件位置。| |*SSL verify host*|勾选此复选框以验证 web 服务器证书的 *Common Name* 字段或 *Subject Alternate Name* 字段是否匹配。
这会设置 [`CURLOPT_SSL_VERIFYHOST`](http://curl.haxx.se/libcurl/c/CURLOPT_SSL_VERIFYHOST.html) cURL 选项。| |*SSL certificate file*|用于客户端认证的 SSL 证书文件名。证书文件必须为 PEM^1^ 格式。支持用户宏。
如果证书文件中也包含私钥，请将 *SSL key file* 字段留空。如果密钥已加密，请在 *SSL key password* 字段中指定密码。包含此文件的目录由 Zabbix 服务器或 proxy 配置参数 [`SSLCertLocation`](/manual/appendix/config/zabbix_server#sslcertlocation) 指定。| |*SSL key file*|用于客户端认证的 SSL 私钥文件名。私钥文件必须为 PEM^1^ 格式。支持用户宏。
包含此文件的目录由 Zabbix 服务器或 proxy 配置参数 [`SSLKeyLocation`](/manual/appendix/config/zabbix_server#sslkeylocation) 指定。| |*SSL key password*|SSL 私钥文件密码。支持用户宏。| |*Description*|输入 connector 描述。| |*Enabled*|勾选此复选框以启用 connector。| 当 Kafka connector 配置了以逗号分隔的 bootstrap broker 地址列表时（例如 `Kafka.URL=kafka1.example.com:9093,kafka2.example.com:9093`），Kafka 客户端会连接到最先响应的 broker，并使用其集群元数据。 如果列表中包含来自不同 Kafka 集群的地址，则只会使用响应最快的集群，其他地址会被记录为不可用；因此，即使 connector 已连接，启动时仍可能出现如下警告： ```default kafka cluster connected, but broker(s) "kafka1.example.com:9093, kafka2.example.com:9093" unavailable; will retry on message send if active brokers fail ``` 在某些环境中（私有网络、容器网络或非标准 DNS/hosts 配置），主机名或 IP 可能会解析为回环地址（例如 `127.0.0.1`/`localhost`），或者被客户端规范化，这会使此类警告具有误导性。 为减少混淆，请确保所有 `Kafka.URL` 地址都属于同一个 Kafka 集群，验证 connector 主机上的 DNS 解析以及 broker 的 `advertised.listeners`，并优先使用可解析为 broker 广播地址的地址。 [comment]: # ({/695ba0a3-cfad3c2f}) [comment]: # ({3208b284-b67df609}) #### 协议 服务器与接收器之间的通信是通过 HTTP 使用 REST API、NDJSON（"Content-Type: application/x-ndjson"）进行的。 有关更多详细信息，请参阅 [Newline-delimited JSON export protocol](/manual/appendix/protocols/real_time_export)。 [comment]: # ({/3208b284-b67df609}) [comment]: # ({fd0d393d-ae5d9204}) ##### 服务器请求 流式传输 监控项值 的示例： ```json POST /v1/history HTTP/1.1 Host: localhost:8080 Accept: */* Accept-Encoding: deflate, gzip, br, zstd Content-Length: 628 Content-Type: application/x-ndjson {"host":{"host":"Zabbix server","name":"Zabbix server"},"groups":["Zabbix servers"],"item_tags":[{"tag":"foo","value":"test"}],"itemid":44457,"name":"foo","clock":1673454303,"ns":800155804,"value":0,"type":3} {"host":{"host":"Zabbix server","name":"Zabbix server"},"groups":["Zabbix servers"],"item_tags":[{"tag":"foo","value":"test"}],"itemid":44457,"name":"foo","clock":1673454303,"ns":832290669,"value":1,"type":3} {"host":{"host":"Zabbix server","name":"Zabbix server"},"groups":["Zabbix servers"],"item_tags":[{"tag":"bar","value":"test"}],"itemid":44458,"name":"bar","clock":1673454303,"ns":867770366,"value":123,"type":3} ``` 流式传输事件的示例： ```json POST /v1/events HTTP/1.1 Host: localhost:8080 Accept: */* Accept-Encoding: deflate, gzip, br, zstd Content-Length: 333 Content-Type: application/x-ndjson {"clock":1673454303,"ns":800155804,"value":1,"eventid":5,"name":"trigger for foo being 0","severity":0,"hosts":[{"host":"Zabbix server","name":"Zabbix server"}],"groups":["Zabbix servers"],"tags":[{"tag":"foo_trig","value":"test"},{"tag":"foo","value":"test"}]} {"clock":1673454303,"ns":832290669,"value":0,"eventid":6,"p_eventid":5} ``` [comment]: # ({/fd0d393d-ae5d9204}) [comment]: # ({d6988482-89101d44}) ##### 接收方响应 响应由 HTTP 响应状态码和 JSON 负载组成。 对于处理成功的请求，HTTP 响应状态码必须为 "200"、"201"、"202"、"203" 或 "204"；对于处理失败的请求，则为其他状态码。 成功示例： ```json HTTP/1.1 200 OK Content-Type: application/json X-Content-Type-Options: nosniff Date: Tue, 21 Apr 2026 10:13:04 GMT Content-Length: 23 {"response":"success"} ``` 带有错误的示例： ```json HTTP/1.1 422 Unprocessable Entity Content-Type: application/json X-Content-Type-Options: nosniff Date: Tue, 21 Apr 2026 12:15:01 GMT Content-Length: 55 {"error":"invalid character '{' after top-level value"} ``` [comment]: # ({/d6988482-89101d44})