从零开始搭建Kafka集群遇到的问题

Posted 2022-04-01 lolxxs

文章目录

• 安装kafka
• • 下载kafka_2.12-3.1.0.tgz
  • 使用Xshell将文件传入虚拟机
  • 解压
• 启动kafka
• • 使用kakfa自带的zookeeper
  • 修改kafka配置
  • 启动kafka
• 使用kafka
• • 创建主题
  • 查看主题
  • 生产消息
  • 消费消息
• 搭建kafka集群
• • 搭建zookeeper集群
  • • 修改配置
    • 启动zookeeper集群
  • 搭建kafka集群
  • • 修改配置
    • 启动kafka集群
• 使用kafka集群
• • 生产消息
  • 消费消息

安装kafka

下载kafka_2.12-3.1.0.tgz

链接：https://pan.baidu.com/s/1TWROqaFLmqaXPcEYzUUQuQ
提取码：kx2o

下载得到kafka_2.11-0.11.0.0.tgz文件

使用Xshell将文件传入虚拟机

使用sftp协议，主机ip是在命令行输入ifconfig中ens33中的ip地址，第一次需要输入虚拟机登录账户root和密码

解压

tar -xzvf kafka_2.11-0.11.0.0.tgz
mv kafka_2.11-0.11.0.0 /opt/kafka

启动kafka

使用kakfa自带的zookeeper

根据zookeeper.properties配置文件启动zookeeper

cd /opt/kafka
/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties

修改kafka配置

vi /opt/kafka/config/server.properties

只有listeners和default.replication.factor需要配置，其他的都有默认配置

broker.id=0 #kafka broker的编号，集群里每个broker的id需不同
listeners=PLAINTEXT://localhost:9092 #listeners是监听地址
log.dirs=/tmp/kafka-logs #日志目录
num.partitions=1 # 新建Topic的默认Partition数量
default.replication.factor=3 #kafka保存消息的副本数

启动kafka

再开一个终端terminal，用于启动kafka，因为没有使用后台启动，如果zookeeper和kafka需要后台启动加个-daemon参数即可如，/opt/kafka/bin/zookeeper-server-start.sh -deamon /opt/kafka/config/zookeeper.properties

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

出现如下情况则启动成功

使用kafka

再创建一个终端用于使用kafka

创建主题

/opt/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic mytopic

查看主题

/opt/kafka/bin/kafka-topics.sh --list --zookeeper  localhost:2181

生产消息

运行后即可在命令行输入字符串，即要发送的消息

/opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic mytopic

消费消息

需要再开一个终端用于接收消息

/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mytopic --from-beginning

搭建kafka集群

由于只使用一个虚拟机，所以需要搭建伪集群，所以先将文件复制两次，即得到三个kafka文件夹

cp -a /opt/kafka  /opt/kafka1
cp -a /opt/kafka  /opt/kafka2

搭建zookeeper集群

要搭建kafka集群之前必须搭建zookeeper集群，先修改zookeeper.properties配置文件

修改配置

/opt/kafka/zookeeper.properties

vi /opt/kafka/config/zookeeper.properties

修改clientPort，并在末尾添加server.1、server.2、server.3

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/tmp/zookeeper
clientPort=2281
server.1=127.0.0.1:2886:3886
server.2=127.0.0.1:2888:3888
server.3=127.0.0.1:2889:3889

/opt/kafka1/zookeeper.properties

vi /opt/kafka1/config/zookeeper.properties

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/tmp/zookeeper1
clientPort=2281
server.1=127.0.0.1:2886:3886
server.2=127.0.0.1:2888:3888
server.3=127.0.0.1:2889:3889

/opt/kafka2/zookeeper.properties

vi /opt/kafka2/config/zookeeper.properties

修改clientPort，并在末尾添加server.1、server.2、server.3

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/tmp/zookeeper2
clientPort=2381
server.1=127.0.0.1:2886:3886
server.2=127.0.0.1:2888:3888
server.3=127.0.0.1:2889:3889

修改完配置还需要创建myid文件，对应server.1、server.2、server.3配置，myid文件在必须dataDir配置的目录下

mkdir  /tmp/zookeeper
mkdir  /tmp/zookeeper1
mkdir  /tmp/zookeeper2

touch /tmp/zookeeper/myid
touch /tmp/zookeeper1/myid
touch /tmp/zookeeper2/myid

然后将myid文件内容分别改为1、2、3

vi /tmp/zookeeper/myid
vi /tmp/zookeeper1/myid
vi /tmp/zookeeper2/myid

使用netstat检查要开启的九个端口有没有被占用，如果被占用了则需要修改配置

netstat -anp | grep 2181
netstat -anp | grep 2281
netstat -anp | grep 2381
netstat -anp | grep 2886
netstat -anp | grep 3886
netstat -anp | grep 2888
netstat -anp | grep 3888
netstat -anp | grep 2889 
netstat -anp | grep 3889

启动zookeeper集群

创建三个端口，启动之前需要进入管理员权限，或者普通用户被授予可执行权限

/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties
/opt/kafka1/bin/zookeeper-server-start.sh /opt/kafka1/config/zookeeper.properties
/opt/kafka2/bin/zookeeper-server-start.sh /opt/kafka2/config/zookeeper.properties

如果出现如下信息，且无报错，则启动集群成功，一般启动一个是会报错的，必须要全部启动，集群才不会报错

搭建kafka集群

修改配置

/opt/kafka/config/server.properties

vi /opt/kafka/config/server.properties

修改配置为如下，其他配置不要改动

broker.id=0
listeners=PLAINTEXT://localhost:9092
log.dirs=/tmp/kafka0-logs
zookeeper.connect=localhost:2181,localhost:2281,localhost:2381

需要自己创建/tmp/kafka0-logs文件夹

mkdir /tmp/kafka0-logs

/opt/kafka1/config/server.properties

vi /opt/kafka1/config/server.properties

修改配置为如下，其他配置不要改动

broker.id=1
listeners=PLAINTEXT://localhost:9093
log.dirs=/tmp/kafka1-logs
zookeeper.connect=localhost:2181,localhost:2281,localhost:2381

需要自己创建/tmp/kafka1-logs文件夹

mkdir /tmp/kafka1-logs

/opt/kafka2/config/server.properties

vi /opt/kafka2/config/server.properties

修改配置为如下，其他配置不要改动

broker.id=2
listeners=PLAINTEXT://localhost:9094
log.dirs=/tmp/kafka2-logs
zookeeper.connect=localhost:2181,localhost:2281,localhost:2381

需要自己创建/tmp/kafka2-logs文件夹

mkdir /tmp/kafka2-logs

启动kafka集群

创建三个终端，启动集群，也是需要权限

/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
/opt/kafka1/bin/kafka-server-start.sh /opt/kafka1/config/server.properties
/opt/kafka2/bin/kafka-server-start.sh /opt/kafka2/config/server.properties

出现如下情况，说明我们启动的三个服务器kafaka集群的id分别是0、1、2，且已经启动成功

使用kafka集群

生产消息

运行后即可在命令行输入字符串，即要发送的消息

/opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic mytopic

消费消息

需要再开一个终端用于接收消息，可以看到，发送消息的端口是9092，但是消费消息的端口是9093，9094，也仍然可以消费消息

/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic mytopic --from-beginning
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 --topic mytopic --from-beginning