目录

1、准备工作

2、搭建zookeeper集群

3、搭建kafka集群

4、测试kafka集群

4.1 命令测试

4.2 java代码测试

1、准备工作

kafka-3.2.0的单机安装教程如下：

CentOS7.9安装kafka-3.2.0和window10 下安装kafka-3.2.0_fyihdg的博客-CSDN博客CentOS7.9安装kafka-3.2.0https://blog.csdn.net/fyihdg/article/details/125552717

在这个文章基础上搭建的集群，我的是三台服务器。

192.168.1.61

192.168.1.62

192.168.1.63

软件包已经上传到了服务器：

修改主机名 

 vi /etc/hosts

 我三台的hosts文件都是修改成如下

192.168.1.61 node01
192.168.1.62 node02
192.168.1.63 node03

 java 环境变更也早配置好了

安装时间同步插件：

yum install ntp -y

 同步服务器时间命令：

ntpdate ntp1.aliyun.com

clock -w

 2、搭建zookeeper集群

  修改zk配置文件,zoo.cfg

vi zoo.cfg

修改192.168.1.61的服务器，修改的地方有：

dataDir=/root/zkdata

 与单机的区别就是要添加以下信息：

server.1=node01:2888:3888
server.2=node02:2888:3888
server.3=node03:2888:3888

记得创建目录：

mkdir /root/zkdata

 剩下的两台都是相同的操作。

修改myid

 192.168.1.61上执行：
 echo 1 > /root/zkdata/myid

 192.168.1.62上执行：
 echo 2 > /root/zkdata/myid


 192.168.1.63上执行：
 echo 3 > /root/zkdata/myid

 

 

 切换到目录：/opt/apache-zookeeper-3.8.0/bin，三台服务器执行，

./zkServer.sh start zoo.cfg

 然后查看状态：

 

 可以看到，63这台是主节点，zookeeper 集群搭建好了。

3、搭建kafka集群

切换到kafa文件目录，然后解压：

 tar -zxf kafka_2.13-3.2.0.tgz -C /opt

切换到目录：/opt/kafka_2.13-3.2.0/config， 

修改配置文件： server.properties，192.168.1.61修改如下：

listeners=PLAINTEXT://node01:9092

log.dirs=/usr/kafka-logs

zookeeper.connect=node01:2181,node02:2181,node03:2181

修改配置文件： server.properties，192.168.1.62修改如下：

broker.id=1

listeners=PLAINTEXT://node02:9092

log.dirs=/usr/kafka-logs

zookeeper.connect=node01:2181,node02:2181,node03:2181

修改配置文件： server.properties，192.168.1.63修改如下：

broker.id=2

listeners=PLAINTEXT://node03:9092

log.dirs=/usr/kafka-logs

zookeeper.connect=node01:2181,node02:2181,node03:2181

配置就完成了

4、测试kafka集群

4.1 命令测试

 启动kafka服务：

3台机器分别执行以下命令： 

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

创建主题：

./bin/kafka-topics.sh  --bootstrap-server node01:9092,node02:9092,node03:9092  --create  --topic topic01  --partitions 3 --replication-factor 2

 如果三台机器都创建主题的话，会报错：“Error while executing topic command : Topic 'topic02' already exists.”

查看有多少队列：

./bin/kafka-topics.sh 	  --bootstrap-server node01:9092,node02:9092,node03:9092 --list

查看主题详细信息：

./bin/kafka-topics.sh  --bootstrap-server node01:9092,node02:9092,node03:9092  --describe   --topic topic01

看一下/usr/kafka-logs这个目录有什么内容

 如有需要可以修改：

./bin/kafka-topics.sh  --bootstrap-server node01:9092,node02:9092,node03:9092   --alter  --topic topic02  --partitions 3

删除：

./bin/kafka-topics.sh     --bootstrap-server node01:9092,node02:9092,node03:9092   --delete   --topic topic03

订阅

输入命令：

./bin/kafka-console-consumer.sh  --bootstrap-server node01:9092,node02:9092,node03:9092  --topic topic01  --group g1 --property print.key=true --property print.value=true  --property key.separator=,

 生产者：

./bin/kafka-console-producer.sh   --broker-list node01:9092,node02:9092,node03:9092   --topic topic01

 

 

     消费组

  ./bin/kafka-consumer-groups.sh  --bootstrap-server node01:9092,node02:9092,node03:9092    --list  g1

消费组详细

 ./bin/kafka-consumer-groups.sh  --bootstrap-server node01:9092,node02:9092,node03:9092     --describe   --group g1

4.2 java代码测试

 如果是window 10,要修改hosts文件，首先到目录：C:WindowsSystem32driversetc

然后修改hosts文件，添加集群的主机映射：

192.168.1.61    node01 
192.168.1.62    node02 
192.168.1.63    node03 

 

 新建代码：

public class KafkaConsumerDemo {
    public static void main(String[] args) {
        //1.创建Kafka链接参数
        Properties props=new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node01:9092,node02:9092,node03: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,"group01");

        //2.创建Topic消费者
        KafkaConsumer consumer=new KafkaConsumer(props);
        //3.订阅topic开头的消息队列
        consumer.subscribe(Pattern.compile("^topic.*$"));

        while (true){
            ConsumerRecords consumerRecords = consumer.poll(Duration.ofSeconds(1));
            Iterator> recordIterator = consumerRecords.iterator();
            while (recordIterator.hasNext()){
                ConsumerRecord record = recordIterator.next();
                String key = record.key();
                String value = record.value();
                long offset = record.offset();
                int partition = record.partition();
                System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset);
            }
        }
    }
}

pom.xml:

    4.0.0

    com.hdg
    kafka
    1.0-SNAPSHOT
    
        
            
                org.apache.maven.plugins
                maven-compiler-plugin
                
                    8
                    8
                
            
        
    


    

            





            
                org.apache.kafka
                kafka-streams
                2.2.0
            

            
            
                log4j
                log4j
                1.2.17
            
            
            
                org.slf4j
                slf4j-api
                1.7.25
            
            
            
                org.slf4j
                slf4j-log4j12
                1.7.25
            

            
            
                org.apache.commons
                commons-lang3
                3.9
            


    

 如果不配置hosts文件，可能会报：

 

 到此集群搭建成功。