Kafka：轻松安装ZooKeeper和Kafka，开启消息传递之旅

2023-05-17 10:54:32

踏上Kafka之旅：下载、安装、初探与提示

踏入消息传递的广阔世界，Apache Kafka是一个不可或缺的工具，它提供了可靠、可扩展且高吞吐量的消息传递平台。让我们携手踏上Kafka之旅，从下载和安装到初探其核心功能，最后分享一些贴心提示，助你开启一段激动人心的消息传递之旅。

下载与安装

ZooKeeper：Kafka的基石

ZooKeeper是一个分布式协调服务，它是Kafka的支柱。首先，从官方网站下载ZooKeeper并将其解压到你的计算机上。接下来，编辑conf/zoo.cfg文件，设置dataDir和clientPort等基本配置。最后，启动ZooKeeper服务器：

zkServer.sh start

Kafka下载与解压

前往Apache Kafka官网下载最新版本的Kafka并解压到你的计算机上。编辑conf/server.properties文件，配置重要的参数，如broker.id、listeners和zookeeper.connect。根据需要，还可以修改conf/log4j.properties文件中的日志级别。

启动Kafka集群

准备好所有配置后，启动Kafka服务器：

kafka-server-start.sh server.properties

稍等片刻，Kafka服务器将启动并运行。

初探Kafka世界

创建主题

主题是Kafka中消息的逻辑分组。使用kafka-topics.sh脚本创建主题：

kafka-topics.sh --create --topic test --partitions 3 --replication-factor 2

其中，--partitions指定分区数，--replication-factor指定副本数。

编写生产者代码

生产者将消息发送到Kafka主题。编写一个Java生产者代码，使用KafkaProducer发送消息：

// ...
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class MyProducer {

  public static void main(String[] args) {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

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

    ProducerRecord<String, String> record = new ProducerRecord<>("test", "Hello, Kafka!");

    producer.send(record);

    producer.close();
  }
}

编写消费者代码

消费者从Kafka主题中读取消息。编写一个Java消费者代码，使用KafkaConsumer从主题中消费消息：

// ...
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class MyConsumer {

  public static void main(String[] args) {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

    consumer.subscribe(Collections.singletonList("test"));

    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

      for (ConsumerRecord<String, String> record : records) {
        System.out.println("Received message: " + record.value());
      }
    }

    consumer.close();
  }
}