Kafka Driver 部署指南

介绍

Kafka Driver 是 oslo.messaging 中的一个实验性消息传输后端。该驱动程序将 oslo.messaging 的基础通知消息交换能力映射到 Apache Kafka 分布式流平台的 v2.0 版本。有关 Apache Kafka 服务器的更多详细信息，请访问 Apache Kafka 网站。

有关驱动程序实现的更多详细信息，请访问 添加 kafka 驱动程序规范 和 更新 kafka 驱动程序规范。

概述

Kafka 驱动程序仅支持用于发送和接收 oslo.messaging 通知。具体而言，Kafka 驱动程序不支持 oslo.messaging RPC 传输。驱动程序与 Kafka 服务器后端之间的通信使用 通过 TCP 的二进制协议，该协议将所有 API 定义为请求响应消息对。Kafka 驱动程序集成了 confluent-kafka Python 客户端以提供完整的协议支持，并利用 Producer API 发布通知消息，利用 Consumer API 进行通知监听器订阅。该驱动程序可以与单个 Kafka 服务器实例或集群 Kafka 服务器部署一起工作。

混合消息部署

Oslo.messaging 提供了一种机制，可以为 RPC 和通知通信配置单独的消息后端。这通过在服务配置中定义单独的 RPC 和通知 传输 URL 来支持。当 Kafka 驱动程序部署用于 oslo.messaging 通知时，必须为 RPC 通信部署单独的驱动程序和消息后端。对于这些混合消息配置，可以部署 rabbit 驱动程序用于 oslo.messaging RPC。

Topics 和 vhost 支持

Kafka topic 是发布记录的 feed 名称。Kafka 中的 topic 是多订阅者，这意味着一个 topic 可以有零个、一个或多个订阅其写入数据的消费者。在 oslo.messaging 中，一个通知监听器订阅一个 topic，该 topic 在提供的目标中直接映射到 Kafka topic。Kafka 服务器架构本身不支持 vhost。为了支持驱动程序提供的传输 URL 中 vhost 的存在，Kafka 服务器上创建的 topic 将附加虚拟主机名称。这会为每个虚拟主机创建一个唯一的 topic，但请注意 Kafka 服务器本身不提供任何访问控制或隔离。

监听器池

Kafka 驱动程序支持监听器池。此功能通过将监听器池名称映射到 Kafka 服务器consumer group 名称来实现。发布到 topic 的每条记录都将被传递到每个订阅池（例如，consumer group）中的一个消费者实例。如果未将监听器池名称分配给通知监听器，Kafka 驱动程序将使用单个默认consumer group，并且所有监听器都将被分配到该组，消息将在竞争的监听器实例之间有效地进行负载均衡。

同步提交

Kafka 服务器和经典 broker 队列之间的一个主要功能差异在于，从 commit log 读取的消息的偏移量或位置由监听器（例如，消费者）控制。驱动程序将线性地推进其维护的偏移量，因为它从服务器读取消息记录。为了确保在停机或通信中断期间不会生成重复的消息，驱动程序将在通知监听器分发之前同步提交消耗的消息。由于此原因，驱动程序不支持重新排队操作，并且驱动程序无法从 Kafka 分区重放消息。

先决条件

为了运行驱动程序，必须安装 confluent-kafka Python 客户端。Kafka 驱动程序集成了基于 librdkafka 的 Python 客户端，以提供完整的协议支持，并利用 Producer API 发布通知消息，利用 Consumer API 进行通知监听器订阅。

可以通过 PyPI 获取 confluent-kafka 库的源代码包。由于 Kafka 驱动程序是 oslo.messaging 的可选扩展，因此这些包默认情况下未安装。在安装 oslo.messaging 时使用 kafka extras 标签，以拉取这些额外的包

$ python -m pip install oslo.messaging[kafka]

配置

传输 URL 启用

在 oslo.messaging 中，transport_url 参数定义了 RPC 和 Notify 的 OpenStack 服务后端。URL 的形式如下

transport://user:pass@host1:port[,hostN:portN]/virtual_host

其中 transport 值指定 RPC 或通知后端，例如 rabbit、kafka 等。要指定并启用通知的 Kafka 驱动程序，请在服务配置文件的 [oslo_messaging_notifications] 部分中指定 transport_url 参数

[oslo_messaging_notifications]
transport_url = kafka://username:password@kafkahostname:9092

请注意，如果在 [oslo_messaging_notifications] 部分中未指定 transport_url 参数，则 [DEFAULT] transport_url 的值将用于 RPC 和通知后端。

驱动程序选项

建议使用 Kafka 驱动程序提供的默认配置选项。可以在服务配置文件的 oslo_messaging_kafka 部分中修改配置选项。

通知监听器选项

• oslo_messaging_kafka.kafka_max_fetch_bytes

• oslo_messaging_kafka.kafka_consumer_timeout

• oslo_messaging_kafka.consumer_group

• oslo_messaging_kafka.enable_auto_commit

• oslo_messaging_kafka.max_poll_records

Notifier 选项

• oslo_messaging_kafka.producer_batch_timeout

• oslo_messaging_kafka.producer_batch_size

compression_codec

由 producer 生成的所有数据的压缩编解码器，有效值为：none、gzip、snappy、lz4、zstd。请注意，此选项的合法性取决于 kafka 版本，请参阅 kafka 文档。

安全选项

• oslo_messaging_kafka.security_protocol

• oslo_messaging_kafka.sasl_mechanism

• oslo_messaging_kafka.ssl_cafile

DevStack 支持

Kafka oslo.messaging 驱动程序的插件受 DevStack 支持。由于 Kafka 驱动程序只能部署用于通知，因此该插件支持部署几种消息总线配置。在 [localrc] 部分的 local.conf 中，必须启用 devstack-plugin-kafka 插件仓库。例如

[[local|localrc]]
enable_plugin kafka https://opendev.org/openstack/devstack-plugin-kafka

如果需要，设置 Kafka 和 Scala 版本和位置变量

KAFKA_VERSION=2.0.0
KAFKA_BASEURL=http://archive.apache.org/dist/kafka
SCALA_VERSION=2.12
SCALA_BASEURL=https://scala-lang.org.cn/riles/archive

RPC_ 和 NOTIFY_ 变量将定义将使用的消息总线配置。混合配置将允许使用 rabbit 驱动程序用于 RPC 传输，而 kafka 驱动程序将用于通知传输。服务变量的设置将选择为配置启用哪个消息中间件

	RPC		NOTIFY
	SERVICE	PORT	SERVICE	PORT
配置 1	rabbit	5672	kafka	9092