目录
kafka-3.2.0的单机安装教程如下：
在这个文章基础上搭建的集群，我的是三台服务器。
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 node03java 环境变更也早配置好了
安装时间同步插件：
yum install ntp -y
同步服务器时间命令：
ntpdate ntp1.aliyun.com
clock -w
修改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 集群搭建好了。
切换到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
配置就完成了
启动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
如果是window 10,要修改hosts文件，首先到目录：C:\Windows\System32\drivers\etc
然后修改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<String,String> consumer=new KafkaConsumer<String, String>(props);
//3.订阅topic开头的消息队列
consumer.subscribe(Pattern.compile("^topic.*$"));
while (true){
ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator();
while (recordIterator.hasNext()){
ConsumerRecord<String, String> 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:
<?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.hdg</groupId>
<artifactId>kafka</artifactId>
<version>1.0-SNAPSHOT</version>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>8</source>
<target>8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<!-- <dependency>-->
<!-- <groupId>org.apache.kafka</groupId>-->
<!-- <artifactId>kafka-clients</artifactId>-->
<!-- <version>2.2.0</version>-->
<!-- </dependency>-->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>2.2.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/log4j/log4j -->
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.25</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.25</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.9</version>
</dependency>
</dependencies>
</project>
如果不配置hosts文件，可能会报：
到此集群搭建成功。
下载地址：
官网地址：EFAK
我已经把安装包上传到CSDN，免积分可以下载：
kafka-eagle-bin-2.1.0.tar.gz-Java文档类资源-CSDN下载kafka-eagle-bin-2.1.0.tar.gz2022年7月份下载，最新版更多下载资源、学习资料请访问CSDN下载频道.https://download.csdn.net/download/fyihdg/85901920上传到服务器，我是安装在192.168.1.61这一台服务器，只在一台安装：
解压到 /opt
https://github.com/smartloli/EFAKtar -xvf kafka-eagle-bin-2.1.0.tar.gz -C /opt 
可以看到， 还是一个压缩包，到/opt/kafka-eagle-bin-2.1.0目录，再解压：
tar -xvf efak-web-2.1.0-bin.tar.gz 解压后：
移到到 /opt目录
mv efak-web-2.1.0 /opt
把源安装包删除
rm -rf kafka-eagle-bin-2.1.0/ 
修改环境变量：
vi /etc/profile
使文件生效
source /etc/profile切换到目录：/opt/efak-web-2.1.0/conf，修改配置文件：system-config.properties
efak.zk.cluster.alias=cluster1
cluster1.zk.list=node01:2181,node02:2181,node03:2181
#cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12:2181
cluster1.efak.offset.storage=kafka
#cluster2.efak.offset.storage=zk
#cluster2.efak.sasl.enable=false
#cluster2.efak.sasl.protocol=SASL_PLAINTEXT
#cluster2.efak.sasl.mechanism=PLAIN
#cluster2.efak.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-eagle";
#cluster2.efak.sasl.client.id=
#cluster2.efak.blacklist.topics=
#cluster2.efak.sasl.cgroup.enable=false
#cluster2.efak.sasl.cgroup.topics=
######################################
# kafka ssl authenticate
######################################
#cluster3.efak.ssl.enable=false
#cluster3.efak.ssl.protocol=SSL
#cluster3.efak.ssl.truststore.location=
#cluster3.efak.ssl.truststore.password=
#cluster3.efak.ssl.keystore.location=
#cluster3.efak.ssl.keystore.password=
#cluster3.efak.ssl.key.password=
#cluster3.efak.ssl.endpoint.identification.algorithm=https
#cluster3.efak.blacklist.topics=
#cluster3.efak.ssl.cgroup.enable=false
#cluster3.efak.ssl.cgroup.topics=
#这个是我window mysql
efak.driver=com.mysql.cj.jdbc.Driver
efak.url=jdbc:mysql://192.168.1.66:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
efak.username=root
efak.password=root切换到目录：/opt/kafka_2.13-3.2.0/bin，3台服务器都要修改,添加：
vi kafka-server-start.sh
export JMX_PORT="7788"
3台服务器配置好后，重启一下
./kafka-server-stop.sh
./bin/kafka-server-start.sh -daemon config/server.properties切换到目录：/opt/efak-web-2.1.0/bin，查看，发现没有执行权限：
授权：
chmod 777 *启动：
./ke.sh start 
如果不能连接，比如报以下错误：
发现没开启远程访问
解决办法：
在装有MySQL的机器上登录MySQL mysql -u root -p密码
执行use mysql;
执行update user set host = '%' where user = 'root';这一句执行完可能会报错，不用管它。
执行FLUSH PRIVILEGES;
经过上面4步，就可以解决这个问题了。
输入访问地址：http://192.168.1.61:8048
* Account:admin ,Password:123456
如果发现无法登录，可以查看日志：/opt/efak-web-2.1.0/logs
less log.log比如说，会报：出现The server time zone value ‘�й���ʱ��’ is unrecogni异常
mysql连接加上&serverTimezone=UTC
又发现报错了
### The error occurred while setting parameters
### SQL: insert into ke_metrics(`cluster`,`broker`,`type`,`key`,`value`,`timespan`,`tm`) values (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?) , (?,?,?,?,?,?,?)
### Cause: java.sql.SQLSyntaxErrorException: Table 'ke.ke_metrics' doesn't exist
; bad SQL grammar []; nested exception is java.sql.SQLSyntaxErrorException: Table 'ke.ke_metrics' doesn't exist这是因为mysql8.0不支持，我们得使用mysql5.7版本！！
自动创建数据库和表了：
搭建成功了，一定要注意，不能使用mysql8.0，否则是无法进入的
