随着云计算和微服务架构的兴起，使用 Kubernetes 管理应用程序的趋势越来越明显。对于容器化的应用程序而言，消息传递是一种常见的通信模式，而 Kafka 则是一款流行的分布式消息传递系统。在 Kubernetes 集群中使用 Kafka 传递消息不仅可以提高应用程序的可靠性和灵活性，还可以使应用程序更易于扩展和部署。本文将介绍如何在 Kubernetes 上使用 Kafka 传递消息，并提供示例代码和指导意义。
为什么选择 Kafka
Kafka 是一款由 Apache 软件基金会开发的分布式流处理平台，具有高可靠性、高可扩展性和低延迟的特点。Kafka 支持发布/订阅和点对点通信模式，并能够处理海量的数据流。在微服务架构中，各个服务之间需要进行信息交换，而 Kafka 可以提供可靠的消息传递服务，同时也具备高性能和容错性能。在 Kubernetes 集群中使用 Kafka 可以使应用程序更易于部署和管理，同时也减少了应用程序之间的紧耦合关系，提高了系统的可维护性和可扩展性。
在 Kubernetes 中部署 Kafka
要在 Kubernetes 中部署 Kafka，首先需要创建一个 ZooKeeper 服务，因为 Kafka 是依赖于 ZooKeeper 的。在 Kubernetes 中，可以使用 StatefulSet 来部署 ZooKeeper，如下所示：
-- -------------------- ---- -------
----------- -------
----- -----------
---------
----- ---------
-----
------------ ---------
--------- -
---------
------------
---- ---------
---------
---------
-------
---- ---------
-----
-----------
- ----- ---------
------ ---------------
------
- ----- ---------
-------------- ----
- ----- ---------
-------------- ----
- ----- -----------
-------------- ----
-------------
- ----- ----
---------- -----
----
- ----- ---------
----------
---------
---------- -------------
- ----- -----------
------ -------------------------------------------------------------------------------------------------
--------
- ----- ----
--------- --上述配置文件中，我们指定了一个名为 zookeeper 的 StatefulSet，并指定了它的副本数量为 3。同时，我们也定义了它的容器模板，使用 zookeeper:3.6.0 镜像来运行 ZooKeeper。这里的环境变量 ZOO_MY_ID 的值将会自动与它的 pod 名称匹配。在 volumes 字段中，我们还创建了一个名为 data 的存储卷，用于存储 ZooKeeper 的数据。最后，ZOO_SERVERS 环境变量指定了每个节点的网络地址和端口号。
接下来，我们创建一个 Kafka StatefulSet，如下所示：
-- -------------------- ---- -------
----------- -------
----- -----------
---------
----- -----
-----
------------ -----
--------- -
---------
------------
---- -----
---------
---------
-------
---- -----
-----
-----------
- ----- -----
------ -----------------------------
------
- ----- ------------
-------------- ----
-------------
- ----- ----
---------- -----
----
- ----- ---------------
----------
---------
---------- -------------
- ----- -----------------------
------ --------------
- ----- --------------------------
------ ----------------------- ---------
- ----- --------------------------------------
------ ---
- ----- -------------------------
------ ------
--------
- ----- ----
--------- --上述配置文件中，我们指定了一个名为 kafka 的 StatefulSet，并指定了它的副本数量为 3。同时，我们也定义了它的容器模板，使用 wurstmeister/kafka:2.12-2.5.0 镜像来运行 Kafka。在 env 字段中，我们指定了 Kafka 的各项配置属性，如 Broker ID、ZooKeeper 连接地址、广播地址等等。注意到我们将 KAFKA_ADVERTISED_LISTENERS 环境变量值设置为 $(hostname -f):9092，这样 Kafka 就会使用当前 pod 的主机名来广播其监听地址。最后，我们在 volumes 字段中创建了一个名为 data 的存储卷，用于存储 Kafka 的数据。
通过上面的配置，我们已经在 Kubernetes 集群中成功部署了一个 Kafka 集群，现在就可以开始使用它进行消息传递了。
在应用程序中使用 Kafka
在应用程序中使用 Kafka 进行消息传递，需要使用 Kafka 客户端库来调用 Kafka 的 API。在 Java 应用程序中，可以使用 Kafka 的 Java 客户端库，如下所示：
-- -------------------- ---- -------
------ ------------------------------------
------ ------------------------------------
------ ----------------------------------------
------ ---------------------
------ ----- ------------ -
------- ------ ----- ------ ----- - -----------
------- ------ ----- ------ ----------------- - -------------
------ ------ ---- -------------- ----- ------ --------- -
--------------
--------------
-
------- ------ -------------- ------- ---------------- -
---------- ----- - --- -------------
-------------------------------------------------- -------------------
----------------------------------------------------- --------------------------------
------------------------------------------------------- ----------------------------------
------ --- -----------------------
-
------- ------ -------------- ------- --------------------- -------- -
---------- ----- - --- -------------
-------------------------------------------------- -------------------
----------------------------------------- ---------
------------------------------------------------------- ----------------------------------
--------------------------------------------------------- ------------------------------------
-------------------------------------------------- ------------
----- -------------- ------- -------- - --- -----------------------
-----------------------------------------------------
------ ---------
-
------ ---- ------------- -
----- -------------- ------- -------- - -----------------
--- ----- ----- - -- ----- - --- -------- -
----- -------------------- ------- ------ - --- ----------------------- ------ ------ - - -------
----- -------- -------- - ---------- ---------- -- -
-- ---------- -- ----- -
-------------------------- -- ---- -------- - - ------------------------
- ---- -
------------------------ ------- -- - ------------ - -- - - - ---------------- - --- - -------------------- - --- - -------------------
-
--
--------------------- ----------
-
-----------------
-----------------
-
------ ---- ------------- ------ -------------------- -
----- -------------- ------- -------- - ---------------------------
----- --- ------ - ----
--- -------------- - --
----- ------ -
----- --------------------- ------- --------------- - ---------------------------------------
-- ------------------------ -- -- -
-----------------
-- --------------- - ------- ------
---- ---------
-
------------------------------ -- -
--------------------------- ------- ---- - -------- ---- --- -- --- ----- --------
------------- --------------- ------------------- ---------------- ------------- ----------------
---
-----------------------
-
-----------------
-
-上述代码中，我们使用了 Kafka 的 Java 客户端库，并创建了一个 KafkaExample 类。该类包含 runProducer() 和 runConsumer() 两个函数，用于发送和接收消息。
在 createProducer() 函数中，我们使用了 KafkaProducer 类来创建一个生产者。通过设置 BOOTSTRAP_SERVERS_CONFIG、KEY_SERIALIZER_CLASS_CONFIG 和 VALUE_SERIALIZER_CLASS_CONFIG 共三个属性，我们可以指定 Kafka 的连接地址、消息键的序列化器和消息值的序列化器。
在 runProducer() 函数中，我们使用了 for 循环向 Kafka 中发送了 10 条消息。在 createConsumer() 函数中，我们使用了 KafkaConsumer 类来创建一个消费者。通过设置 BOOTSTRAP_SERVERS_CONFIG、GROUP_ID_CONFIG、KEY_DESERIALIZER_CLASS_CONFIG、VALUE_DESERIALIZER_CLASS_CONFIG 和 AUTO_OFFSET_RESET_CONFIG 共五个属性，我们可以指定 Kafka 的连接地址、消费组 ID、消息键的反序列化器、消息值的反序列化器和自动重置偏移量的策略。
在 runConsumer() 函数中，我们使用了 while 循环来不断轮询 Kafka 中是否有消息需要消费。通过调用 poll() 函数，我们可以从 Kafka 中获取消息。在获取到消息之后，我们可以通过 for 循环遍历所有消息，并打印出它们的键和值。
在 Kubernetes 中部署应用程序
在 Kubernetes 中部署应用程序，我们需要编写一个部署文件，来定义要部署的容器。例如，下面是一个使用 Java Spring Boot 框架的部署文件示例：
-- -------------------- ---- -------
----------- -------
----- ----------
---------
----- ------
-----
--------- -
---------
------------
---- ------
---------
---------
-------
---- ------
-----
-----------
- ----- ------
------ -------------------------------
------
- ----- ----
-------------- ----
----
- ----- ------------------------------
------ ----------上述代码中，我们指定了一个名为 my-app 的 Deployment，并将它的容器副本数量设置为 1。同时，我们指定容器的镜像路径、端口以及其他必要的配置属性，在 SPRING_KAFKA_BOOTSTRAP_SERVERS 变量中，我们指定了 Kafka 的连接地址。
当你将上述部署文件应用到 Kubernetes 集群中时，Kubernetes 将创建一个名为 my-app 的 Deployment，并启动一个容器副本。该容器将使用指定的镜像路径和配置，启动一个 Spring Boot 应用程序。应用程序将会自动连接到 Kafka，并使用上面的 Kafka Java 客户端库发送和接收消息了。
总结
在本文中，我们介绍了如何在 Kubernetes 上使用 Kafka 进行消息传递，并提供了详细的部署流程和示例代码。通过使用 Kafka，可以使微服务架构更加灵活和可靠，同时也可以提高系统的可维护性和可扩展性。如果您还没有使用 Kafka 进行消息传递，那么现在就可以开始尝试了！
来源：JavaScript中文网 ，转载请注明来源 https://www.javascriptcn.com/post/647aa09a968c7c53b0650964