快速入门、安装

消息队列介绍

消息队列包括两种模式，点对点模式和发布/订阅模式

1. 点对点模式

在点对点消息系统中，消息持久化到一个队列中。此时，将有一个或多个消费者消费队列中的数据。但是一条消息只能被消费一次。当一个消费者消费了队列中的某条数据之后，该条数据则从消息队列中删除。该模式即使有多个消费者同时消费数据，也能保证数据处理的顺序【总结：生产者发送一条消息到queue，只有一个消费者能收到】。

2. 发布/订阅模式

在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者【总结：发布者发送到topic的消息，只有订阅了topic的订阅者才会收到消息。kafka即采用该模式】。

什么是kafka

Kafka 起初是由 Linkedin 公司采用 Scala 语言开发的一个多分区、多副本且基于ZooKeeper协调的分布式消息系统，现己被捐献给 Apache 基金会。目前 Kafka 已经定位为一个分布式流式处理平台，它以高吞吐、可持久化、可水平扩展、支持流数据处理等多种特性而被广泛使用。

kafka的架构

Kafak 总体架构图中包含多个概念：

Zookeeper： zookeeper负责保存broker集群元数据，并对控制器进行选举等操作。
Broker： 一个独立的kafka服务器被称作broker。
1. broker负责接收来自生产者的消息，为消息设置偏移量，并将消息存储在磁盘。
2. broker为消费者提供服务，对读取分区的请求作出响应，返回已经提交到磁盘上的消息。
Producer： 生产者负责创建消息，将消息发送到broker。
Consumer： 消费者负责从Broker订阅并消费消息。
Consumer Group： Consumer Group 为消费者组，一个消费者可以包含一个或多个 Consumer
使用 多分区 + 多消费者 方式可以极大 提高数据下游的处理速度，同一消费者组中的消费者不会重复消费消息，同样的，不同消费组中的消费者消费消息时互不影响。
Topic： Kafka 中的消息 以 Topic 为单位进行划分，生产者将消息发送到特定的 Topic，而消费者负责订阅 Topic 的消息并进行消费。
Partition： 一个 Topic 可以细分为多个分区，每个分区只属于单个主题。同一个主题下不同分区包含的消息是不同的，分区在存储层面可以看作一个可追加的 日志（Log）文件，消息在被追加到分区日志文件的时候都会分配一个特定的 偏移量（offset）。
Offset： offset 是消息在分区中的唯一标识，Kafka 通过它来保证消息在分区内的顺序性，不过 offset 并不跨越分区，也就是说，Kafka保证的是分区有序性而不是主题有序性。
Replication： 是 Kafka 保证数据高可用的方式，Kafka 同一 Partition 的数据可以在多 Broker 上存在多个副本，通常只有主副本对外提供读写服务，当主副本所在 broker 崩溃或发生网络异常，Kafka 会在 Controller 的管理下会重新选择新的 Leader 副本对外提供读写服务。
Record： 实际写入 Kafka 中并可以被读取的消息记录。每个 record 包含了 key、value 和 timestamp。
Leader： 每个分区多个副本的 “主” leader,生产者发送数据的对象，以及消费者消费数据的对象都是 leader。
Follower： 每个分区多个副本中的"从" follower,实时从 Leader 中同步数据，保持和 leader 数据的同步。Leader 发生故障时，某个 follow 会成为新的 leader。

kafka重要组件

1. Topic（主题）

kafka将消息以topic为单位进行归类；
topic始终是支持多用户订阅的，也就是说，一个主题可以有零个，一个或者多个消费者订阅写入的数据；
在kafka集群中，可以有无数的topic；
生产者和消费者消费数据一般以topic为单位。更细粒度可以到paritition级别。

2. Partition（分区）

一个topic可以有多个分区（partition），每个分区保存部分topic的数据，所有的partition当中的数据全部合并起来，就是一个topic当中的所有的数据；
在kafka中，每一个分区会有一个编号：编号从0开始。每一个分区内的数据是有序的，但全局的数据不能保证是有序的。（有序是指生产什么样顺序，消费时也是什么样的顺序）

3. Consumer Group（消费者组）

消费者组由一个或者多个消费者组成，同一个组中的消费者对于同一条消息只消费一次。
每个消费者都属于某个消费者组，如果不指定，那么所有的消费者都属于默认的组。
每个消费者组都有一个ID，即group ID。组内的所有消费者协调在一起来消费一个订阅主题( topic)的所有分区(partition)。当然，每个分区只能由同一个消费组内的一个消费者(consumer)来消费，可以由不同的消费组来消费。
partition数量决定了每个consumer group中并发消费者的最大数量，如下图：

如上面左图所示，如果只有两个分区，即使一个组内的消费者有4个，也会有两个空闲的。
如上面右图所示，有4个分区，每个消费者消费一个分区，并发量达到最大4。
再来看如下一幅图：

如上图所示，不同的消费者组消费同一个topic，这个topic有4个分区，分布在两个节点上。左边的消费组1有两个消费者，每个消费者就要消费两个分区才能把消息完整的消费完，右边的消费组2有四个消费者，每个消费者消费一个分区即可。

总结下kafka中分区与消费组的关系：

消费组：由一个或者多个消费者组成，同一个组中的消费者对于同一条消息只消费一次。某一个主题下的分区数，对于消费该主题的同一个消费组下的消费者数量，应该小于等于该主题下的分区数。

如：某一个主题有4个分区，那么消费组中的消费者应该小于等于4，而且最好与分区数成整数倍 1 2 4 这样。同一个分区下的数据，在同一时刻，不能同一个消费组的不同消费者消费。

总结：分区数越多，同一时间可以有越多的消费者来进行消费，消费数据的速度就会越快，提高消费的性能。

4. Partition replicas （分区副本）

副本数（replication-factor）：控制消息保存在几个broker（服务器）上，一般情况下副本数等于broker的个数。
一个broker服务下，不可以创建多个副本因子。创建Topic时，副本因子应该小于等于可用的broker数。副本因子操作以partition为单位的。每个分区都有各自的主副本和从副本；
主副本叫做leader，从副本叫做 follower（在有多个副本的情况下，kafka会为同一个分区下的所有分区，设定角色关系：一个leader和N个 follower），处于同步状态的副本叫做in-sync-replicas(ISR);
消费者和生产者都是从leader读写数据，不与follower交互。follower通过pull的方式从leader同步数据，尽量让自己跟leader副本的内容一致。

副本因子的作用：冗余备份。让kafka读取数据和写入数据时的可靠性。

5. Segment文件

一个partition当中由多个segment文件组成，每个segment文件，包含两部分，一个是 .log 文件，另外一个是 .index 文件，其中 .log 文件包含了我们发送的数据存储，.index 文件，记录的是我们.log文件的数据索引值，以便于我们加快数据的查询速度。

索引文件与数据文件的关系

上图左半部分是索引文件，里面存储的是一对一对的key-value，其中key是消息在数据文件（对应的log文件）中的编号，比如“1,3,6,8……”，分别表示在log文件中的第1条消息、第3条消息、第6条消息、第8条消息……

value 代表的是在全局partiton中的第几个消息。

那么为什么在index文件中这些编号不是连续的呢？这是因为index文件中并没有为数据文件中的每条消息都建立索引，而是采用了稀疏存储的方式，每隔一定字节的数据建立一条索引。这样避免了索引文件占用过多的空间，从而可以将索引文件保留在内存中。但缺点是没有建立索引的Message也不能一次定位到其在数据文件的位置，从而需要做一次顺序扫描，但是这次顺序扫描的范围就很小了。

稀疏索引：为了数据创建索引，但范围并不是为每一条创建，而是为某一个区间创建；好处：就是可以减少索引值的数量。不好的地方：找到索引区间之后，要得进行第二次处理。

segment文件命名规则：partion全局的第一个segment从0开始，后续每个segment文件名为上一个全局 partion的最大offset（偏移message数）。数值最大为64位long大小，20位数字字符长度，没有数字就用 0 填充。

kafka安装（windows版）

1. 安装Zookeeper

地址：http://zookeeper.apache.org/releases.html#download
解压并进入Zookeeper目录下conf文件夹，将【zoo_sample.cfg】 => 【zoo.cfg】
打开【zoo.cfg】找到并编辑dataDir = D:\zookeeper\data ；dataLogDir=D:\zookeeper\log（视安装目录更改）
添加系统变量：
打开新的cmd，输入“zkServer”，运行Zookeeper
命令行提示如下，说明本地Zookeeper启动成功

2. 安装kafka

下载安装包：
kafka_2.12-2.3.0-SNAPSHOT

官网下载的版本貌似是有BUG的，运行一段时间后kafka会自动清理过期日志，在清理过程中会报出【该文件有另一程序正在使用】的错误，导致kafka停止运行
好在国外大牛将该bug整合处理后发了新包，版本为：【kafka_2.12-2.3.0-SNAPSHOT】

目录结构：
修改配置
进入config目录找到文件server.properties并打开。找到并编辑log.dirs=.\kafka\logs；zookeeper.connect=127.0.0.1:2181
kafka会按照默认，在9092端口上运行，并连接zookeeper的默认端口：2181
双击该目录下kafka-start.bat执行或者打开cmd进入kafka目录，输入：

.\bin\windows\kafka-server-start.bat .\config\server.properties

附：server.properties配置文件说明：

#broker的全局唯一编号，不能重复
broker.id=0

#用来监听链接的端口，producer或consumer将在此端口建立连接
port=9092

#处理网络请求的线程数量
num.network.threads=3

#用来处理磁盘IO的线程数量
num.io.threads=8

#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400

#接受套接字的缓冲区大小
socket.receive.buffer.bytes=102400

#请求套接字的缓冲区大小
socket.request.max.bytes=104857600

#kafka运行日志存放的路径
log.dirs=/export/data/kafka/

#topic在当前broker上的分片个数
num.partitions=2

#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1

#segment文件保留的最长时间，超时将被删除
log.retention.hours=168

#滚动生成新的segment文件的最大时间
log.roll.hours=1

#日志文件中每个segment的大小，默认为1G
log.segment.bytes=1073741824

#周期性检查文件大小的时间
log.retention.check.interval.ms=300000

#日志清理是否打开
log.cleaner.enable=true

#broker需要使用zookeeper保存meta数据
zookeeper.connect=zk01:2181,zk02:2181,zk03:2181

#zookeeper链接超时时间
zookeeper.connection.timeout.ms=6000

#partion buffer中，消息的条数达到阈值，将触发flush到磁盘
log.flush.interval.messages=10000

#消息buffer的时间，达到阈值，将触发flush到磁盘
log.flush.interval.ms=3000

#删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除
delete.topic.enable=true

#此处的host.name为本机IP(重要),如果不改,则客户端会抛出:Producer connection to localhost:9092 unsuccessful 错误!
host.name=kafka01

advertised.host.name=192.168.140.128

producer生产者配置文件说明
#指定kafka节点列表，用于获取metadata，不必全部指定
metadata.broker.list=node01:9092,node02:9092,node03:9092
# 指定分区处理类。默认kafka.producer.DefaultPartitioner，表通过key哈希到对应分区
#partitioner.class=kafka.producer.DefaultPartitioner
# 是否压缩，默认0表示不压缩，1表示用gzip压缩，2表示用snappy压缩。压缩后消息中会有头来指明消息压缩类型，故在消费者端消息解压是透明的无需指定。
compression.codec=none
# 指定序列化处理类
serializer.class=kafka.serializer.DefaultEncoder
# 如果要压缩消息，这里指定哪些topic要压缩消息，默认empty，表示不压缩。
#compressed.topics=

# 设置发送数据是否需要服务端的反馈,有三个值0,1,-1
# 0: producer不会等待broker发送ack 
# 1: 当leader接收到消息之后发送ack 
# -1: 当所有的follower都同步消息成功后发送ack. 
request.required.acks=0 

# 在向producer发送ack之前,broker允许等待的最大时间 ，如果超时,broker将会向producer发送一个error ACK.意味着上一次消息因为某种原因未能成功(比如follower未能同步成功) 
request.timeout.ms=10000

# 同步还是异步发送消息，默认“sync”表同步，"async"表异步。异步可以提高发送吞吐量,
也意味着消息将会在本地buffer中,并适时批量发送，但是也可能导致丢失未发送过去的消息
producer.type=sync

# 在async模式下,当message被缓存的时间超过此值后,将会批量发送给broker,默认为5000ms
# 此值和batch.num.messages协同工作.
queue.buffering.max.ms = 5000

# 在async模式下,producer端允许buffer的最大消息量
# 无论如何,producer都无法尽快的将消息发送给broker,从而导致消息在producer端大量沉积
# 此时,如果消息的条数达到阀值,将会导致producer端阻塞或者消息被抛弃，默认为10000
queue.buffering.max.messages=20000

# 如果是异步，指定每次批量发送数据量，默认为200
batch.num.messages=500

# 当消息在producer端沉积的条数达到"queue.buffering.max.meesages"后 
# 阻塞一定时间后,队列仍然没有enqueue(producer仍然没有发送出任何消息) 
# 此时producer可以继续阻塞或者将消息抛弃,此timeout值用于控制"阻塞"的时间 
# -1: 无阻塞超时限制,消息不会被抛弃 
# 0:立即清空队列,消息被抛弃 
queue.enqueue.timeout.ms=-1


# 当producer接收到error ACK,或者没有接收到ACK时,允许消息重发的次数 
# 因为broker并没有完整的机制来避免消息重复,所以当网络异常时(比如ACK丢失) 
# 有可能导致broker接收到重复的消息,默认值为3.
message.send.max.retries=3

# producer刷新topic metada的时间间隔,producer需要知道partition leader的位置,以及当前topic的情况 
# 因此producer需要一个机制来获取最新的metadata,当producer遇到特定错误时,将会立即刷新 
# (比如topic失效,partition丢失,leader失效等),此外也可以通过此参数来配置额外的刷新机制，默认值600000 
topic.metadata.refresh.interval.ms=60000

3. 测试是否搭建成功

【模拟生产者创建topic】：打开cmd窗口进入kafka目录，输入：

.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
PS：创建名称为“test”的topic

【模拟生产者发送消息】：打开cmd窗口进入kafka目录，输入：

.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test

【模拟消费者消费消息】：额外打开cmd窗口进入kafka目录，输入：

.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning

在步骤2的cmd窗口输入消息，步骤3的cmd窗口实时收取到消息即代表大功告成。

原理分析

生产者

生产消息分区分配策略

生产者数据不丢失机制

ack机制（确认机制）

生产者数据发送出去，需要服务端返回一个确认码，即ack响应码；ack的响应有三个状态值0,1，-1

0：生产者只负责发送数据，不关心数据是否丢失，丢失的数据，需要再次发送

1：partition的leader收到数据，不管follow是否同步完数据，响应的状态码为1

-1：所有的从节点都收到数据，响应的状态码为-1

如果broker端一直不返回ack状态，producer永远不知道是否成功；producer可以设置一个超时时间10s，超过时间认为失败。

发送消息方式

生产者发送给kafka数据，可以采用 同步方式 或 异步方式

同步方式：

发送一批数据给kafka后，等待kafka返回结果：

生产者等待10s，如果broker没有给出ack响应，就认为失败。
生产者重试3次，如果还没有响应，就报错.

异步方式：

发送一批数据给kafka，只是提供一个回调函数：

先将数据保存在生产者端的buffer中。buffer大小是2万条。
满足数据阈值或者数量阈值其中的一个条件就可以发送数据。
发送一批数据的大小是500条。

如果broker迟迟不给ack，而buffer又满了，开发者可以设置是否直接清空buffer中的数据。

消费者

消费组分配策略

提交消费

重平衡机制（rebalance）

原文链接：https://blog.csdn.net/zhangleiQAQ/article/details/122614438

目录