准备三台互通服务器（关闭防火墙）
192.168.203.136
192.168.203.137
192.168.203.138
一.搭建zookeeper集群
1、在/usr/local目录下解压安装包，
tar -zxvf apache-zookeeper-3.7.1-bin.tar.gz
链接：百度网盘 请输入提取码
提取码：bd2u
2、进入zookeeper目录下新建文件夹zkData
mkdir zkData
3、进入conf目录下修改配置文件
mv zoo_sample.cfg zoo.cfg
vi zoo.cfg
(修改dataDir参数)
dataDir=/usr/local/zookeeper-3.7.1/zkData
(新增下面三行,2888端口用于主从信息同步,3888端口用于选举)
server.1=192.168.203.136:2888:3888
server.2=192.168.203.138:2888:3888
server.3=192.168.203.137:2888:3888
保存退出
4、进入zkData目录，将myid文件上传到该目录下（myid中的数字与上面server.1 2 3表示的服务器对应）
5、安装剩余的两台服务器（重复上面操作，myid要修改为对应的数字）
6、依次进入bin目录启动三台zookeeper
./zkServer.sh start (启动)
./zkServer.sh status (查看状态，一主两从即为成功)
二.搭建kafka集群
链接：百度网盘 请输入提取码
提取码：q4pm
1、在/usr/local目录下解压安装包
tar -zxvf kafka_2.12-3.3.2.tgz
2、进入kafka/config/目录下修改配置文件
vi server.properties（broker.id 不能相同，可以依次设置为0，1，2。ip改为对应服务器的ip地址）
broker.id=0（修改参数）
port=9092 （新增参数）
host.name=192.168.203.136 （新增参数）
steners=PLAINTEXT://192.168.203.136:9092 （新增参数）
advertised.listeners=PLAINTEXT://192.168.203.136:9092 （新增参数）
zookeeper.connect=192.168.203.136:2181,192.168.203.138:2181,192.168.203.137:2181（修改参数）
3、安装剩余的两台服务器（重复上面操作，修改broker.id）
4、依次进入kafka目录下，启动kafka
./bin/kafka-server-start.sh -daemon ./config/server.properties &
netstat -tunlp |grep 9092 （检查kafka端口）
注1：kafak新版本操作命令
启动命令：
bin/kafka-server-start.sh -daemon config/server.properties
创建topic
./kafka-topics.sh --create --bootstrap-server 192.168.203.136:9092 --replication-factor 1 --partitions 1 --topic test
查看topic
./kafka-topics.sh --bootstrap-server 192.168.203.136:9092 --list
向指定topic中生产数据
./kafka-console-producer.sh --broker-list 192.168.203.136:9092 --topic test
例如：{"id":"1","name":"xiaoming","age":"20"}
查看topic具体内容
./kafka-console-consumer.sh --bootstrap-server 192.168.203.136:9092 --topic test --from-beginning
创建消费者组
./kafka-console-consumer.sh --bootstrap-server 192.168.203.136:9092 --topic test --group consumer-10
查看消费者组
./kafka-consumer-groups.sh --bootstrap-server 192.168.203.136:9092 --list
查看消费者详情
./kafka-consumer-groups.sh --bootstrap-server 192.168.203.136:9092 --describe --group consumer-10
消费数据
./kafka-console-consumer.sh --bootstrap-server 192.168.203.136:9092 --topic test --from-beginning
注2：kafka集群监控系统安装
1、选择集群中任意一台服务器上传并解压安装包
unzip kafka-manager-2.0.0.2.zip
2、进入kafka-manager/conf目录下修改配置文件
vi application.conf
kafka-manager.zkhosts="192.168.203.136:2181" (修改参数ip为zookeeper主机ip)
3、返回kafka-manager目录启动系统
nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=9000 >kafka-manager.log 2>&1 &
浏览器访问ip:9000端口即可。
注3：java连接kafka集群：
1、pom文件增加依赖
<!-- kafka -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>2、修改配置文件（groupId自定义）
kafka:
bootstrapServers: 192.168.203.136:9092,192.168.203.138:9092,192.168.203.137:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
groupId: consumer-10
auto-offset-reset: latest
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
enable-auto-commit: false
auto-commit-interval: 5000
max-poll-records: 13、生产者文章来源：https://www.toymoban.com/news/detail-429595.html
@Autowired
private KafkaTemplate kafkaTemplate;
@Scheduled(cron = "0/5 * * * * ?")
public void kafkaTest(){
kafkaTemplate.send("test","测试kafka+时间为"+LocalDateTime.now());
log.info("发送kafka test 数据：");
}4、消费者文章来源地址https://www.toymoban.com/news/detail-429595.html
package com.example.demo.controller;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class KafkaListener1 {
@KafkaListener(topics = "test")
public void listenKafkaHour1(String payMessage) {
log.info("收到kafka test 数据：" + payMessage);
}
}
到了这里，关于kafka集群搭建的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
