检测本地JDK版本,保证为JDK1.8版本

#检测JDK版本
rpm -qa | grep openjdk

# 卸载旧版JDK
yum -y remove java-1.7.0-openjdk-headless-1.7.0.111-2.6.7.8.el7.x86_64

# 安装JDK1.8
yum install java-1.8.0-openjdk.x86_64

查看是否启用zookeeper

安装zookeeper方法参考 Zookeeper搭建（集群）

netstat -antpl | grep 2181

下载kafka

官网地址：http://kafka.apache.org/

下载地址：https://www.apache.org/dyn/closer.cgi?path=/kafka/2.2.0/kafka_2.12-2.2.0.tgz

cd /opt/kafka
wget https://www-eu.apache.org/dist/kafka/2.2.0/kafka_2.12-2.2.0.tgz

单机部署

修改配置信息

cd /opt/kafka/kafka_2.12-2.2.0

# 创建日志目录
mkdir -p /opt/kafka/kafka_2.12-2.2.0/logs

# 修改配置，参考下方说明
vim config/server.properties 

# 启动服务
./bin/kafka-server-start.sh -daemon /opt/kafka/kafka_2.12-2.2.0/config/server.properties 

注：下面使用的zookeeper与kafka为同一台服务器部署

# 测试是否生效
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
显示：Created topic test.

# 查看topic分区
./bin/kafka-topics.sh --list --zookeeper localhost:2181

# 生产消息并且消费
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>hello !
>hello wecom kafka!
>hello my name githen
>
#每一次回车换行表示一条消息，使用ctrl+c结束生产消息


消费者消费消息命令
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning // 支持旧版本

注：以下为server.properties部分配置信息的说明

#一个broker在集群中的唯一标示，要求是正数。在改变IP地址，不改变broker.id的话不会影响consumers
broker.id=1

# 若使用1024以下端口，需要全用root权限启动
#listeners=PLAINTEXT://:9092
 
#advertised.listeners=PLAINTEXT://your.host.name:9092
 
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SS
 
# broker 处理消息的最大线程数，一般情况下不需要去修改
num.network.threads=3
 
# broker处理磁盘IO 的线程数 ，数值应该大于你的硬盘数
num.io.threads=8
 
# socket的发送缓冲区（SO_SNDBUF）
socket.send.buffer.bytes=102400
 
# socket的接收缓冲区 (SO_RCVBUF)
socket.receive.buffer.bytes=102400
 
# 压缩后限制单个消息的大小，默认值1000000,也就是1MB
message.max.bytes

# socket请求的最大字节数。为了防止内存溢出，message.max.bytes必然要小于
socket.request.max.bytes=104857600
 
#kafka数据的存放地址，多个地址的话用逗号分割 /tmp/kafka-logs-1，/tmp/kafka-logs-2，按最少使用原则
# 把同一个分区 的日志片段保存到同一个路径下。
# 要注意：broker会往拥有最少数目分区的路径新增分区，而不是往拥有最小磁盘空间的路径新增分区
log.dirs=/opt/kafka/kafka_2.12-2.2.0/logs
 
# 每个topic的分区个数，更多的partition会产生更多的segment file
# 自动创建主题，主题的个数即为num.partitions的数
# 主题分区的个数不能减少，只能增加
# 哪果要让一个主题的分区少于num.partitions的值，需要手动创建主题
num.partitions=1

# 如下三种情况，kafka会使用可配置的线程池来处理日志片片段
# * 1. 服务器正常启动，用于打开 每个分区的日志片段
# * 2. 服务器崩溃后重启，用户检查和截短每个分区 的日志片段
# * 3. 服务器正常关闭，用于关闭日志片段
# 试例：num.recovery.threads.per.data.dir设为8，并且log.dir指定3个路径，那共需要24个线程
num.recovery.threads.per.data.dir=1
 
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
 
# 当达到下面的消息数量时，会将数据flush到日志文件中。默认10000
#log.flush.interval.messages=10000
 
# 当达到下面的时间(ms)时，执行一次强制的flush操作。interval.ms和interval.messages无论哪个达到，都会flush。默认3000ms
#log.flush.interval.ms=1000
 
# 日志保存时间 (hours|minutes)，默认为7天（168小时）。超过这个时间会根据policy处理数据。bytes和minutes无论哪个先达到都会触发。
# 还有 log.retention.minutes log.retention.ms 推荐使用log.retention.ms
log.retention.hours=168

# 日志保存最大字节数，若一个包含8个分区的主题，且log.retention.bytes设置为1GB,那主题最多可以保留8GB的数据
# 同时设置了log.retention.ms 和 log.retention.bytes，满足任意一条，消息都会被 删除
#log.retention.bytes=1073741824
 
# 控制日志片段文件的大小，超出该大小则追加到一个新的日志segment文件中（-1表示没有限制）
# 这个参数的值越小，就会越频繁地关闭和分配新文件，从而降低磁盘写入的整体效率
# 如一个主题每天接收100MB的消息，而log.segment.bytes使用默认设置(1GB)，那需要10天才能填满一个日志片段，
# 因为日志片段被关闭之前消息是不会过期的，所以如果log.retention.ms设置为604800000(1周)，那日志片段最多需要17天才会过期
# 关闭日志片段需要10天时间，再根据过期时间，还需要再保留7天（要等到日志片段里的最后一个消息过期才能被删除）
log.segment.bytes=1073741824
 

# 日志片段关闭时间，无默认值
# log.segment.ms= 
# 日志片段文件的检查周期，查看它们是否达到了删除策略的设置（log.retention.hours或log.retention.bytes）
log.retention.check.interval.ms=300000
 
# Zookeeper quorum设置。如果有多个使用逗号分割 例如 ip:prot,ip:prot,ip:prot，一旦有一个zk服务器宕机，broker 可以连接到zk的另一个节点上
# 结构如下：hostname:port/path /path 是可选的zookeeper路径，作为 kafka集群的chroot环境，不指定使用根路径，避免多kafka集群的冲突
zookeeper.connect=localhost:2181
 
# 连接zk的超时时间
zookeeper.connection.timeout.ms=6000
 
# ZooKeeper集群中leader和follower之间的同步实际
group.initial.rebalance.delay.ms=0

# kafka会在以下几种情况 创建主题：
# * 1. 当一个生产者开始往主题写入消息时
# * 2. 当一个消费者开始从主题读取消息时
# * 3. 当任意一个客户端向主题发送元数据请求时
# 根据kafka协议，如果 一个主题不先被创建，根本无法知道 它是否已经存在
auto.create.topics.enable=false

集群部署

配置节点

利用单节点部署多个broker模拟集群机制，不同的broker不同的id，监听端口以及日志目录

cd /opt/kafka/kafka_2.12-2.2.0/config

# borker 1
cp server.properties server-1.properties 
vim server-1.properties

> broker.id=1
> listeners=PLAINTEXT://:9091
> log.dirs=/opt/kafka/kafka_2.12-2.2.0/logs/1

cp server.properties server-2.properties 
vim server-2.properties

> broker.id=2
> listeners=PLAINTEXT://:9092
> log.dirs=/opt/kafka/kafka_2.12-2.2.0/logs/2

cp server.properties server-3.properties 
vim server-3.properties

> broker.id=3
> listeners=PLAINTEXT://:9093
> log.dirs=/opt/kafka/kafka_2.12-2.2.0/logs/3


# 启动服务
./bin/kafka-server-start.sh -daemon /opt/kafka/kafka_2.12-2.2.0/config/server-1.properties 
./bin/kafka-server-start.sh -daemon /opt/kafka/kafka_2.12-2.2.0/config/server-2.properties 
./bin/kafka-server-start.sh -daemon /opt/kafka/kafka_2.12-2.2.0/config/server-3.properties 

测试集群

# 添加消息
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic mytest1
#说明：factor：指的使我们副本的数量，这里我们需要三个副本去对应我们的三个broker，让他们自动选举leader。

# 查看topic信息
./bin/kafka-topics.sh --list --zookeeper localhost:2181

# 查看详细信息
./bin/kafka-topics.sh --describe --zookeeper localhost:2181

# 显示内容：
Topic:mytest1 PartitionCount:1 ReplicationFactor:3 Configs:
 Topic: mytest1 Partition: 0 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2
# 说明:leader:主的broker ，replicas:代表副本对应的broker id 号, Isr:代表活着的broker


# 创建消息生产者
./bin/kafka-console-producer.sh --broker-list localhost:9091,localhost:9092,localhost:9093 --topic mytest1

# 消费者消费消息
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9091 --topic mytest1 --from-beginning