centos7安装kafka

1. 安装依赖环境

# 1.1 安装JDK8

解压tar包，放到/usr/local目录下

tar -zvf jdk-8u144-linux-i586.tar.gz
1

编辑profile文件

vim /etc/profile
1

加入下面的内容

export JAVA_HOME=/usr/local/java/jdk1.8.0_144
export JRE_HOME=${JAVA_HOME}/jre 
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib:$CLASSPATH 
export JAVA_PATH=${JAVA_HOME}/bin:${JRE_HOME}/bin 
export PATH=$PATH:${JAVA_PATH}
1
2
3
4
5

# 重新加载配置文件
source /etc/profile
1
2

# 验证结果
java -version
1
2

1.2 安装ZooKeeper

Zookeeper是安装Kafka集群的必要组建，Kafka通过Zookeeper来实施对原数据信息的管理，包括集群、主题、分区等内容。

同样在官网下载安装包到指定目录解压缩，步骤如下：

• Zookeeper官网：https://zookeeper.apache.org/

• 修改Kafka的配置文件，首先进入安装路径conf目录，并将zoo_sample.cfg文件修改为zoo.cfg，并对核心参数进行配置。

mv zoo_sample.cfg zoo.cfg
1

zoo.cfg文件内容如下：

# The number of milliseconds of each tick
# ZK服务器的心跳时间
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
# 投票选取新Leader的初始化时间
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
# 数据目录
dataDir=/opt/apache-zookeeper-3.6.1-bin/data
# 日志目录
dataLogDir=/opt/apache-zookeeper-3.6.1-bin/log
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

这里只需要配置一下dataDir和dataLogDir的目录即可

启动Zookeeper命令：

# 在zookeeper文件目录下执行下面命令
bin/zkServer.sh start
1
2

2. Kafka的安装和配置

官网内容：http://kafka.apache.org/

• 下载解压成功后，开始启动

  启动命令：bin/kafka-server.start.sh config/server.properties

# 在kafka目录下执行
bin/kafka-server.start.sh config/server.properties
1
2

server.properties 需要关注以下几个配置

broker.id=0 表示broker的编号，如果集群中有多个broker，则每个broker的编号需要不同

listeners=PLAINTEXT://:9092 broker对外提供的服务入口地址

advertised.listeners=PLAINTEXT://192.168.152.128:9092

log.dirs=/opt/kafka_2.13-2.5.0/logs 设置存放消息日志文件的地址

zookeeper.connect=localhost:2181 kafka所需zookeeper集群地址。

这里只需要修改的地方是加上下面这两个给开放了，这个原来是注释掉的。

listeners=PLAINTEXT://:9092
# 下面这个IP换成centos的IP地址
dvertised.listeners=PLAINTEXT://192.168.152.128:9092
1
2
3

• 验证结果

[root@localhost bin]# jps -l
54723 sun.tools.jps.Jps
53387 kafka.Kafka
103581 org.apache.zookeeper.server.quorum.QuorumPeerMain
1
2
3
4

2.1 kafka测试消息的生产与消费

2.1.1 首先创建一个主题

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

–zookeeper：制定了kafka所连接的zookeeper服务的地址

–topic：指定了所要创建主题的名称

–partitions：指定了分区个数

–replication-factor：指定了副本因子

–create：创建主题的动作命令

• 展示所有的主题

[root@localhost kafka_2.12-2.5.0]# bin/kafka-topics.sh --zookeeper localhost:2181 --list
__consumer_offsets
liufei
topic-01
yiyang
1
2
3
4
5

• 查看主题详情

[root@localhost kafka_2.12-2.5.0]# bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic yiyang
Topic: yiyang   PartitionCount: 2       ReplicationFactor: 1    Configs: 
        Topic: yiyang   Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: yiyang   Partition: 1    Leader: 0       Replicas: 0     Isr: 0
1
2
3
4

2.1.1 启动消费端接收消息

命令：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic yiyang
1

–bootstrap-server 指定了连接kafka集群的地址

–topic 指定了消费端订阅的主题

2.1.2 启动生产端发送消息

命令：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic yiyang
1

–broker-list 指定连接kafka集群的地址

–topic 指定了发送消息时的主题

这边发了两个消息，dsa和liufei
消费者这边接收到了两个消息