如何在Ubuntu20.04上安装ApacheKafka

作为 Write for DOnations 计划的一部分，作者选择了 Free and Open Source Fund 来接受捐赠。

介绍

Apache Kafka 是一种流行的分布式消息代理，旨在处理大量实时数据。 Kafka 集群具有高度可扩展性和容错性。与 ActiveMQ 和 RabbitMQ 等其他消息代理相比，它还具有更高的吞吐量。虽然它通常用作 发布/订阅 消息传递系统，但许多组织也将它用于日志聚合，因为它为发布的消息提供持久存储。

发布/订阅消息系统允许一个或多个生产者发布消息，而无需考虑消费者的数量或他们将如何处理消息。订阅的客户端会自动收到有关更新和新消息创建的通知。与客户端定期轮询以确定是否有新消息可用的系统相比，该系统更有效且可扩展。

在本教程中，您将在 Ubuntu 20.04 上安装和使用 Apache Kafka 2.6.3。

先决条件

要继续进行，您将需要：

一个 Ubuntu 20.04 服务器和一个具有 sudo 权限的非 root 用户。如果您没有设置非 root 用户，请按照本指南中指定的步骤进行操作。
您的服务器上至少有 4GB 的 RAM。没有此数量的 RAM 的安装可能会导致 Kafka 服务失败。
OpenJDK 11 安装在您的服务器上。要安装此版本，请按照我们的教程如何在 Ubuntu 20.04 上使用 APT 安装 Java。 Kafka 是用 Java 编写的，因此需要 JVM。

第 1 步 — 为 Kafka 创建用户

因为 Kafka 可以通过网络处理请求，所以您的第一步是为该服务创建一个专用用户。如果有人破坏了 Kafka 服务器，这可以最大限度地减少对您的 Ubuntu 机器的损害。我们将在此步骤中创建一个专用的 kafka 用户。

以非 root sudo 用户身份登录，创建一个名为 kafka 的用户：

sudo adduser kafka

按照提示设置密码并创建kafka用户。

接下来，使用 adduser 命令将 kafka 用户添加到 sudo 组。您需要这些权限来安装 Kafka 的依赖项：

sudo adduser kafka sudo

您的 kafka 用户现已准备就绪。使用 su 登录帐户：

su -l kafka

现在您已经创建了一个特定于 Kafka 的用户，您可以下载并提取 Kafka 二进制文件了。

第 2 步 — 下载和提取 Kafka 二进制文件

让我们下载 Kafka 二进制文件并将其解压缩到我们的 kafka 用户主目录中的专用文件夹中。

首先，在 /home/kafka 中创建一个名为 Downloads 的目录来存储您的下载：

mkdir ~/Downloads

使用 curl 下载 Kafka 二进制文件：

curl "https://downloads.apache.org/kafka/2.6.3/kafka_2.13-2.6.3.tgz" -o ~/Downloads/kafka.tgz

创建一个名为 kafka 的目录并切换到该目录。这将是 Kafka 安装的基本目录：

mkdir ~/kafka && cd ~/kafka

使用 tar 命令提取您下载的存档：

tar -xvzf ~/Downloads/kafka.tgz --strip 1

我们指定 --strip 1 标志以确保存档的内容提取到 ~/kafka/ 本身，而不是在其中的另一个目录（例如 ~/kafka/kafka_2.13-2.6.3/）中。

现在我们已经成功下载并解压了二进制文件，我们可以开始配置我们的 Kafka 服务器。

第 3 步 — 配置 Kafka 服务器

Kafka 的默认行为不允许您删除主题。 Kafka 主题是可以向其发布消息的类别、组或提要名称。要修改它，您必须编辑配置文件。

Kafka的配置选项在server.properties中指定。使用 nano 或您喜欢的编辑器打开此文件：

nano ~/kafka/config/server.properties

首先，添加一个允许我们删除 Kafka 主题的设置。将以下内容添加到文件底部：

~/kafka/config/server.properties

delete.topic.enable = true

其次，通过修改logs.dir属性来改变存储Kafka日志的目录：

~/kafka/config/server.properties

log.dirs=/home/kafka/logs

保存并关闭文件。现在您已经配置了 Kafka，下一步是创建 systemd 单元文件，以便在启动时运行和启用 Kafka 服务器。

第 4 步 - 创建 Systemd 单元文件并启动 Kafka 服务器

在本节中，您将为 Kafka 服务创建 systemd 单元文件。这将帮助您以与其他 Linux 服务一致的方式执行常见的服务操作，例如启动、停止和重新启动 Kafka。

Zookeeper 是 Kafka 用来管理其集群状态和配置的服务。它用于许多分布式系统。如果您想了解更多信息，请访问官方 Zookeeper 文档。

为 zookeeper 创建单元文件：

sudo nano /etc/systemd/system/zookeeper.service

在文件中输入以下单位定义：

/etc/systemd/system/zookeeper.service

[Unit]
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
User=kafka
ExecStart=/home/kafka/kafka/bin/zookeeper-server-start.sh /home/kafka/kafka/config/zookeeper.properties
ExecStop=/home/kafka/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

[Unit] 部分指定 Zookeeper 需要网络和文件系统准备好才能启动。

[Service] 部分指定 systemd 应该使用 zookeeper-server-start.sh 和 zookeeper-server-stop.sh shell 文件来启动和停止服务。它还指定了如果 Zookeeper 异常退出，应该重新启动它。

添加此内容后，保存并关闭文件。

接下来，为 kafka 创建 systemd 服务文件：

sudo nano /etc/systemd/system/kafka.service

在文件中输入以下单位定义：

/etc/systemd/system/kafka.service

[Unit]
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple
User=kafka
ExecStart=/bin/sh -c '/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/server.properties > /home/kafka/kafka/kafka.log 2>&1'
ExecStop=/home/kafka/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

[Unit] 部分指定此单元文件依赖于 zookeeper.service。这将确保 zookeeper 在 kafka 服务启动时自动启动。

[Service] 部分指定 systemd 应该使用 kafka-server-start.sh 和 kafka-server-stop.sh shell 文件来启动和停止服务。它还指定如果Kafka异常退出应该重新启动。

现在您已经定义了单位，使用以下命令启动 Kafka：

sudo systemctl start kafka

为确保服务器已成功启动，请检查 kafka 单元的日志日志：

sudo systemctl status kafka

您将收到如下输出：

Output● kafka.service
     Loaded: loaded (/etc/systemd/system/kafka.service; disabled; vendor preset: enabled)
     Active: active (running) since Wed 2021-02-10 00:09:38 UTC; 1min 58s ago
   Main PID: 55828 (sh)
      Tasks: 67 (limit: 4683)
     Memory: 315.8M
     CGroup: /system.slice/kafka.service
             ├─55828 /bin/sh -c /home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/server.properties > /home/kafka/kafka/kafka.log 2>&1
             └─55829 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true -Xlog:gc*:file=>

Feb 10 00:09:38 cart-67461-1 systemd[1]: Started kafka.service.

您现在有一个 Kafka 服务器正在侦听端口 9092。

您已启动 kafka 服务。但是如果你重启你的服务器，Kafka 不会自动重启。要在服务器启动时启用 kafka 服务，请运行以下命令：

sudo systemctl enable zookeeper
sudo systemctl enable kafka

在此步骤中，您启动并启用了 kafka 和 zookeeper 服务。在下一步中，您将检查 Kafka 安装。

第 5 步 — 测试 Kafka 安装

在此步骤中，您将测试您的 Kafka 安装。具体来说，您将发布和使用 “Hello World” 消息，以确保 Kafka 服务器正常运行。

在 Kafka 中发布消息需要：

一个 producer，允许将记录和数据发布到主题。
一个 消费者 ，从主题中读取消息和数据。

首先，创建一个名为 TutorialTopic 的主题：

~/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic TutorialTopic

您可以使用 kafka-console-producer.sh 脚本从命令行创建生产者。它需要 Kafka 服务器的主机名、端口和主题作为参数。

现在将字符串 "Hello, World" 发布到 TutorialTopic 主题：

echo "Hello, World" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TutorialTopic > /dev/null

接下来，使用 kafka-console-consumer.sh 脚本创建一个 Kafka 消费者。它需要 ZooKeeper 服务器的主机名和端口，以及主题名称作为参数。

以下命令使用来自 TutorialTopic 的消息。注意 --from-beginning 标志的使用，它允许消费在消费者启动之前发布的消息：

~/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic TutorialTopic --from-beginning

如果没有配置问题，您将在终端中看到 Hello, World：

OutputHello, World

该脚本将继续运行，等待更多消息发布。要对此进行测试，请打开一个新的终端窗口并登录到您的服务器。

在这个新终端中，启动一个生产者来发布另一条消息：

echo "Hello World from Sammy at DigitalOcean!" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TutorialTopic > /dev/null

您将在消费者的输出中看到此消息：

OutputHello, World
Hello World from Sammy at DigitalOcean!

完成测试后，按 CTRL+C 停止消费者脚本。

您现在已经在 Ubuntu 20.04 上安装并配置了 Kafka 服务器。在下一步中，您将执行一些快速任务来加强 Kafka 服务器的安全性。

第 6 步 — 加固 Kafka 服务器

安装完成后，您可以删除 kafka 用户的管理员权限。在您这样做之前，请注销并以任何其他非 root sudo 用户身份重新登录。如果您仍在运行本教程开始时使用的同一 shell 会话，请键入 exit。

从 sudo 组中删除 kafka 用户：

sudo deluser kafka sudo

为了进一步提高您的 Kafka 服务器的安全性，请使用 passwd 命令锁定 kafka 用户的密码。这确保没有人可以使用此帐户直接登录服务器：

sudo passwd kafka -l

此时，只有 root 或 sudo 用户可以通过键入以下命令以 kafka 身份登录：

sudo su - kafka

以后如果要解锁，请使用 passwd 和 -u 选项：

sudo passwd kafka -u

您现在已成功限制 kafka 用户的管理员权限。您已准备好开始使用 Kafka，或者您可以按照下一个可选步骤将 KafkaT 添加到您的系统中。

第 7 步 — 安装 KafkaT（可选）

KafkaT是Airbnb开发的一个工具。它可以更轻松地查看有关 Kafka 集群的详细信息并从命令行执行某些管理任务。但是因为它是一个 Ruby gem，所以你需要 Ruby 才能使用它。您还需要 build-essential 包来构建 KafkaT 依赖的其他 gem。使用 apt 安装它们：

sudo apt install ruby ruby-dev build-essential

您现在可以使用 gem 命令安装 KafkaT：

sudo CFLAGS=-Wno-error=format-overflow gem install kafkat

在 kafkat 的安装过程中，需要“Wno-error=format-overflow”编译标志来抑制 Zookeeper 的警告和错误。

KafkaT 使用 .kafkatcfg 作为配置文件来确定您的 Kafka 服务器的安装和日志目录。它还应该有一个将 KafkaT 指向 ZooKeeper 实例的条目。

创建一个名为 .kafkatcfg 的新文件：

nano ~/.kafkatcfg

添加以下行以指定有关 Kafka 服务器和 Zookeeper 实例的所需信息：

~/.kafkatcfg

{
  "kafka_path": "~/kafka",
  "log_path": "/home/kafka/logs",
  "zk_path": "localhost:2181"
}

您现在可以使用 KafkaT。首先，您可以使用它来查看有关所有 Kafka 分区的详细信息：

kafkat partitions

您将看到以下输出：

Output[DEPRECATION] The trollop gem has been renamed to optimist and will no longer be supported. Please switch to optimist as soon as possible.
/var/lib/gems/2.7.0/gems/json-1.8.6/lib/json/common.rb:155: warning: Using the last argument as keyword parameters is deprecated
...
Topic                 Partition   Leader      Replicas        ISRs    
TutorialTopic         0             0         [0]             [0]
__consumer_offsets    0               0           [0]                           [0]
...
...

您将看到 TutorialTopic 以及 __consumer_offsets，这是 Kafka 用于存储客户端相关信息的内部主题。您可以放心地忽略以 __consumer_offsets 开头的行。

要了解有关 KafkaT 的更多信息，请参阅其 GitHub 存储库。

结论

现在，Apache Kafka 在您的 Ubuntu 服务器上安全运行。您现在可以使用 Kafka 客户端将 Kafka 集成到您最喜欢的编程语言中。

要了解更多关于 Kafka 的信息，您还可以查阅其文档。