Apache Kafka 是一个开源流处理平台，是一种高吞吐量的分布式发布订阅消息系统，可以处理消费者在网站中的所有动作流数据，其所有功能都是以分布式、高度可扩展、弹性、容错和安全的方式提供的。

在 Apache APISIX 2.14 版本中发布了一个新功能，即提供 Kafka 类型的上游，用户可以通过在路由中配置 scheme 为 Kafka 的上游开启 Kafka 消费者功能，从而在各种环境中实现消息订阅。

本文将介绍 Kafka 发布订阅功能及 kafka-proxy 插件的使用细节，为大家呈现如何将 APISIX 与 Kafka 结合使用在浏览器等连接受限的场景，实现对 Kafka 中消息的消费。

原理

Kafka 使用自定义的 TCP 协议实现 Broker 与消费者之间的通信。在 Apache APISIX 中，你可以通过四层代理实现这部分的代理，但对于终端用户（如浏览器等）与 Broker 通信这种无法直接使用 TCP 连接的场景，则没有办法进行很好的支持。

而现在，客户端可通过 WebSocket 连接到 APISIX，在 APISIX 内部建立与 Kafka 的连接，进而处理客户端的命令（如获取偏移量、获取消息等）。通过 WebSocket 的连接，可以避免在浏览器这种无法直接使用 TCP 连接的场景中从 Kafka 中拉取消息。

通过上述流程图可以看到，这里内部使用自定义的 Protobuf 协议作为通信协议，通过便捷的编译过程，使其后续可在多种语言的程序中使用。在过程中，主要通过 ListOffsets 命令来获取偏移量，通过 Fetch 命令来获取消息。

除此之外，由于 Kafka 消费者的功能是基于 APISIX 的 PubSub 框架，所以你也可以基于该框架扩展其他消息系统，实现更丰富的发布订阅能力。

如何使用

设置 Kafka 路由

APISIX 添加了一种新的上游 Scheme 类型，除支持 HTTP、GRPC 等协议外，现在也支持了 Kafka。在操作过程中，只需将 scheme 字段值设置为 kafka，nodes 字段中设置为 Kafka Broker 的地址与端口，即可开启 APISIX 的 Kafka 发布订阅支持。

curl -X PUT 'http://127.0.0.1:9080/apisix/admin/routes/kafka' \
    -H 'X-API-KEY: ${api-key}' \
    -H 'Content-Type: application/json' \
    -d '{
    "uri": "/kafka",
    "upstream": {
        "nodes": {
            "kafka-server:9092": 1
        },
        "type": "none",
        "scheme": "kafka"
    }
}'
1
2
3
4
5
6
7
8
9
10
11
12
13

（可选）设置 TLS 握手

当 Kafka 上游的 tls 字段存在时，APISIX 会为该连接开启 TLS 握手，同时 tls 中还有一个 verify 字段，你可以通过它控制 TLS 握手时是否校验服务器证书。

curl -X PUT 'http://127.0.0.1:9080/apisix/admin/routes/kafka' \
    -H 'X-API-KEY: ${api-key}' \
    -H 'Content-Type: application/json' \
    -d '{
    "uri": "/kafka",
    "upstream": {
        "nodes": {
            "kafka-server:9092": 1
        },
        "type": "none",
        "scheme": "kafka",
        "tls": {
            "verify": true
        }
    }
}'
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

（可选）设置 SASL/PLAIN 认证

为了支持认证功能，APISIX 中还提供了 kafka-proxy 插件，用户可以通过它为 Kafka 的路由配置 SASL 认证功能，当前它仅支持 PLAIN 模式的认证。

curl -X PUT 'http://127.0.0.1:9080/apisix/admin/routes/kafka' \
    -H 'X-API-KEY: ${api-key}' \
    -H 'Content-Type: application/json' \
    -d '{
    "uri": "/kafka",
    "plugins": {
        "kafka-proxy": {
            "sasl": {
              "username": "user",
              "password": "pwd"
            }
        }
    },
    "upstream": {
        "nodes": {
            "kafka-server:9092": 1
        },
        "type": "none",
        "scheme": "kafka"
    }
}'
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

设置客户端及测试

你可以从 Apache APISIX 的 GitHub 仓库中获取 PubSub 的协议定义，这其中包含 Kafka 的命令与响应，需要将它编译成所需语言的 SDK。

之后便可通过客户端（如浏览器）以 WebSocket 连接之前设置的 Kafka 路由 URI ws://127.0.0.1:9080/kafka，向其发送 PubSubReq 数据。这其中包含需要使用的命令，APISIX 将从 Kafka 中获取数据，并向客户端发送 PubSubResp 响应数据。

如使用完毕，只需要移除路由即可关闭该端点的 Kafka 功能。得益于 APISIX 动态化特性，无需重启即可实现状态更新。

总结

关于 Kafka 发布订阅功能的更多说明和完整配置列表，你可以参考下方链接：

• Kafka 发布订阅功能：https://apisix.apache.org/zh/docs/apisix/next/pubsub/kafka
• kafka-proxy 插件：https://apisix.apache.org/zh/docs/apisix/next/plugins/kafka-proxy

目前，Apache APISIX 也在进行其他消息系统的发布订阅能力支持，如果你对此感兴趣，也欢迎阅读发布订阅（PubSub）模块的开发文档 https://apisix.apache.org/zh/docs/apisix/next/pubsub。

如果你有任何想法，也可以在 GitHub Discussion 中发起讨论，或通过社区邮件列表进行交流。