kafka 集群 ZooKeeper 模式搭建

本文介绍: Ap ac he Kaf k a是一个开源分布式事件流平台，被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序关于Zo oKee per的弃用根据 Kaf ka官网信息，随着Apac he Kaf ka 3.5版本的发布，Zo o keep e r现已被标记为已弃用。未来计划在Apac he Kafka（4.0版）的下一个主要版本中删除Zo oKee per，该版本最快将于2024年4月发布。在弃用阶段，ZooKeeper仍然支持用于Kafka集群元数据的管理，但不建议用于新的部署。

Apache Kafka是一个开源分布式事件流平台，被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序

Kafka 官网：Apache Kafka

关于ZooKeeper的弃用

根据 Kafka官网信息，随着Apache Kafka 3.5版本的发布，Zookeeper现已被标记为已弃用。未来计划在Apache Kafka（4.0版）的下一个主要版本中删除ZooKeeper，该版本最快将于2024年4月发布。在弃用阶段，ZooKeeper仍然支持用于Kafka集群元数据的管理，但不建议用于新的部署。新的部署方式使用 KRaft 模式，KRaft 模式部署可以看笔者的文章《kafka 集群 KRaft 模式搭建》，考虑到一些公司仍然在使用老版本的 Kafka，故笔者写这篇文章记录 Kafka 集群Zookeepe r 模式搭建

官网信息截图

mkdir /usr/local/kafka

tar -xzf kafka_2.13-3.6.0.tgz -C /usr/local/kafka

cd /usr/local/kafka/kafka_2.13-3.6.0/config

vi server.properties

cd /usr/local/kafka/kafka_2.13-3.6.0

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

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

bin/kafka-server-stop.sh

&lt;dependency&gt;
    &lt;groupId&gt;org.apache.kafka</groupId&gt;
    <artifactId&gt;kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
 
    <groupId>com.wsjzzcbq</groupId>
    <artifactId>kafka-learn</artifactId>
    <version>1.0-SNAPSHOT</version>
 
    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.6.0</version>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>11</source>
                    <target>11</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

package com.wsjzzcbq;
 
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
 
/**
 * Demo
 *
 * @author wsjz
 * @date 2023/11/24
 */
public class ProducerDemo {
 
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        //配置集群节点信息
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.3.232:9092,192.168.2.90:9092,192.168.2.11:9092");
        //配置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
 
        Producer<String, String> producer = new KafkaProducer<>(properties);
 
        //topic 名称是demo_topic
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("demo_topic", "明月别枝惊鹊");
        RecordMetadata recordMetadata = producer.send(producerRecord).get();
        System.out.println(recordMetadata.topic());
        System.out.println(recordMetadata.partition());
        System.out.println(recordMetadata.offset());
 
    }
}

package com.wsjzzcbq;
 
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
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.Arrays;
import java.util.Properties;
 
/**
 * ConsumerDemo
 *
 * @author wsjz
 * @date 2023/11/24
 */
public class ConsumerDemo {
 
    public static void main(String[] args) {
        Properties properties = new Properties();
        // 配置集群节点信息
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.3.232:9092,192.168.2.90:9092,192.168.2.11:9092");
 
        // 消费分组名
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "demo_group");
        // 序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);
        // 消费者订阅主题
        consumer.subscribe(Arrays.asList("demo_topic"));
 
        while (true) {
            ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String,String> record:records) {
                System.out.printf("收到消息：partition=%d, offset=%d, key=%s, value=%s%n",record.partition(),
                        record.offset(),record.key(),record.value());
            }
        }
    }
}