云服务器(Linux)安装部署Kafka的详细过程

作为一种分布式消息系统，Kafka 可以快速处理大规模的实时数据。在云服务器中进行 Kafka 的部署和安装，可以更加方便地管理和维护 Kafka 的使用。

1. 安装 Java 环境

由于 Kafka 是基于 Java 编写的，因此在开始安装 Kafka 之前，需要先安装 Java 环境（JDK 8 或以上版本）。可以通过以下命令进行安装：

sudo apt-get update
sudo apt-get install default-jdk

2. 下载和解压 Kafka

在官网下载 Kafka 安装包，比如我们下载版本为 2.8.0 的 Kafka：

wget https://dlcdn.apache.org/kafka/2.8.0/kafka_2.13-2.8.0.tgz

下载完成后，解压到指定目录：

tar -xzvf kafka_2.13-2.8.0.tgz
cd kafka_2.13-2.8.0/

3. 编辑配置文件

Kafka 的配置文件为 config/server.properties ，可以通过编辑该文件进行部分配置：

sudo vi config/server.properties

• 修改 broker.id，注意每个节点的 broker.id 必须不同。

broker.id=0

• 修改 zookeeper.connect，指定 Zookeeper 的地址和端口号。

zookeeper.connect=127.0.0.1:2181

• 修改 listeners，指定 Kafka 监听的 IP 地址和端口号。

listeners=PLAINTEXT://0.0.0.0:9092

4. 启动 Zookeeper

Kafka 的运行依赖于 Zookeeper，因此需要先启动 Zookeeper。

bin/zookeeper-server-start.sh config/zookeeper.properties

5. 启动 Kafka 服务

在 Zookeeper 启动成功后，启动 Kafka 服务。

bin/kafka-server-start.sh config/server.properties

这样就启动完成了 Kafka 服务。可以通过以下命令测试 Kafka 是否正常工作：

bin/kafka-topics.sh --create --topic topic1 --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092
bin/kafka-console-producer.sh --topic topic1 --bootstrap-server localhost:9092
bin/kafka-console-consumer.sh --topic topic1 --from-beginning --bootstrap-server localhost:9092

示例1：使用Python客户端消费Kafka消息

from kafka import KafkaConsumer

consumer = KafkaConsumer('topic1', bootstrap_servers=['localhost:9092'])

for msg in consumer:
    print(msg)

示例2：使用Java客户端生产Kafka消息

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class MyProducer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);

        for(int i = 0; i < 10; i++)
            producer.send(new ProducerRecord<String, String>("topic1", Integer.toString(i), Integer.toString(i)));

        producer.close();

    }

}

以上就是云服务器安装部署 Kafka 的详细过程，基本上可以在云服务器上快速搭建 Kafka 环境，并进行测试。