消息中间件Kafuka学习——初次配置使用

MQ本质：

例如 ActiveMQ、RabbitMQ、RocketMQ 等中间件。采用这种模型，本质就是将要推送的数据，不在存放在当前应用程序的内存中，而是将数据存放到另一个专门负责数据处理的应用程序中，从而实现服务解耦。

kafuka模型:

如果你看不懂这些概念没关系，我会带着大家一起梳理一遍！

• Producer：Producer 即生产者，消息的产生者，是消息的入口
• Broker：Broker 是 kafka 一个实例，每个服务器上有一个或多个 kafka 的实例，简单的理解就是一台 kafka 服务器，kafka cluster表示集群的意思
• Topic：消息的主题，可以理解为消息队列，kafka的数据就保存在topic。在每个 broker 上都可以创建多个 topic 。
• Partition：Topic的分区，每个 topic 可以有多个分区，分区的作用是做负载，提高 kafka 的吞吐量。同一个 topic 在不同的分区的数据是不重复的，partition 的表现形式就是一个一个的文件夹！
• Replication：每一个分区都有多个副本，副本的作用是做备胎，主分区（Leader）会将数据同步到从分区（Follower）。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为 Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本
• Message：每一条发送的消息主体。
• Consumer：消费者，即消息的消费方，是消息的出口。
• Consumer Group：我们可以将多个消费组组成一个消费者组，在 kafka 的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！
• Zookeeper：kafka 集群依赖 zookeeper 来保存集群的的元信息，来保证系统的可用性。
简而言之，kafka 本质就是一个消息系统，与大多数的消息系统一样，主要的特点如下：
• 使用推拉模型将生产者和消费者分离
• 为消息传递系统中的消息数据提供持久性，以允许多个消费者
• 提供高可用集群服务，主从模式，同时支持横向水平扩展
1
2
3
4
5
6
7
8
9
10
11
12
13

tips:

与 ActiveMQ、RabbitMQ、RocketMQ 不同的地方在于，它有一个分区Partition的概念。
这个分区的意思就是说，如果你创建的topic有5个分区，当你一次性向 kafka 中推 1000 条数据时，这 1000 条数据默认会分配到 5 个分区中，其中每个分区存储 200 条数据。
这样做的目的，就是方便消费者从不同的分区拉取数据，假如你启动 5 个线程同时拉取数据，每个线程拉取一个分区，消费速度会非常非常快！
这是 kafka 与其他的消息系统最大的不同！

kafuka安装:

1.安装依赖: 【jdk+zookeeper】

#安装jdk1.8
yum -y install java-1.8.0-openjdk
下载zookeeper，并解压文件包
1
2
3

#在线下载zookeeper
wget http://mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.12/zookeeper-3.4.12.tar.gz

#解压
tar -zxvf zookeeper-3.4.12.tar.gz


#创建数据和日志存放目录
cd /usr/zookeeper/
mkdir data
mkdir log

#把conf下的zoo_sample.cfg备份一份，然后重命名为zoo.cfg
cd conf/
cp zoo_sample.cfg zoo.cfg

配置zookeeper
#编辑zoo.cfg文件
vim zoo.cfg
重新配置dataDir和dataLogDir的存储路径
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

最后，启动 Zookeeper 服务
#进入Zookeeper的bin目录
cd zookeeper/zookeeper-3.4.12/bin

#启动Zookeeper
./zkServer.sh start

#查询Zookeeper状态
./zkServer.sh status

#关闭Zookeeper状态
./zkServer.sh stop
1
2
3
4
5
6
7
8
9
10
11
12

2.正式安装Kafuka

到官网http://kafka.apache.org/downloads.html下载想要的版本，我这里下载是最新稳定版2.8.0。
#下载kafka 安装包
wget https://apache.osuosl.org/kafka/2.8.0/kafka-2.8.0-src.tgz

#解压文件包
tar -xvf kafka-2.8.0-src.tgz

按需修改配置文件server.properties（可选）
#进入配置文件夹
cd  kafka-2.8.0-src/config

#编辑server.properties
vim server.properties

server.properties文件内容如下：
broker.id=0
listeners=PLAINTEXT://localhost:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

其中有四个重要的参数：
• broker.id：唯一标识ID
• listeners=PLAINTEXT://localhost:9092：kafka服务监听地址和端口
• log.dirs：日志存储目录
• zookeeper.connect：指定zookeeper服务地址
可根据自己需求修改对应的配置！
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
28
29
30
31
32
33
34
35
36
37
38
39
40
41

启动kafuka

./kafka-server-start.sh  -daemon ../config/server.properties

1
2

创建主题topic

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic heima --partitions 2 --replication-factor 1

结果：

tips：
topic创建失败可能是因为：
在较新版本（2.2 及更高版本）的 Kafka 不再需要 ZooKeeper 连接字符串，即- -zookeeper localhost:2181。
使用 Kafka Broker的 --bootstrap-server localhost:9092来替代- -zookeeper localhost:2181。

可参考文章