基本介绍和使用
# 基本介绍
Apache Kafka是一个分布式发布 - 订阅消息系统和一个强大的队列,可以处理大量的数据,并使您能够将消息从一个端点传递到另一个端点。 Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上,并在群集内复制以防止数据丢失。 Kafka构建在ZooKeeper同步服务之上。 它与Apache Storm和Spark非常好地集成,用于实时流式数据分析。
# 什么是发布订阅系统
在发布 - 订阅系统中,消息被保留在主题中。 与点对点系统不同,消费者可以订阅一个或多个主题并使用该主题中的所有消息。 在发布 - 订阅系统中,消息生产者称为发布者,消息使用者称为订阅者。 一个现实生活的例子是Dish电视,它发布不同的渠道,如运动,电影,音乐等,任何人都可以订阅自己的频道集,并获得他们订阅的频道时可用。
# kafka的消息传送机制
Kafka 支持 3 种消息投递语义,在业务中,常常都是使用 At least once 的模型。
- At most once:最多一次,消息可能会丢失,但不会重复。
- At least once:最少一次,消息不会丢失,可能会重复。
- Exactly once:只且一次,消息不丢失不重复,只且消费一次。
# kafka集群架构
S.No | 组件和说明 |
---|---|
1 | Broker(代理) Kafka集群通常由多个代理组成以保持负载平衡。 Kafka代理是无状态的,所以他们使用ZooKeeper来维护它们的集群状态。 一个Kafka代理实例可以每秒处理数十万次读取和写入,每个Broker可以处理TB的消息,而没有性能影响。 Kafka经纪人领导选举可以由ZooKeeper完成。 |
2 | ZooKeeper ZooKeeper用于管理和协调Kafka代理。 ZooKeeper服务主要用于通知生产者和消费者Kafka系统中存在任何新代理或Kafka系统中代理失败。 根据Zookeeper接收到关于代理的存在或失败的通知,然后生产者和消费者采取决定并开始与某些其他代理协调他们的任务。 |
3 | **Producers(生产者)**生产者将数据推送给经纪人。 当新代理启动时,所有生产者搜索它并自动向该新代理发送消息。 Kafka生产者不等待来自代理的确认,并且发送消息的速度与代理可以处理的一样快。 |
4 | Consumers(消费者) 因为Kafka代理是无状态的,这意味着消费者必须通过使用分区偏移来维护已经消耗了多少消息。 如果消费者确认特定的消息偏移,则意味着消费者已经消费了所有先前的消息。 消费者向代理发出异步拉取请求,以具有准备好消耗的字节缓冲区。 消费者可以简单地通过提供偏移值来快退或跳到分区中的任何点。 消费者偏移值由ZooKeeper通知。 |
# kafka的工作流程
# 发布 - 订阅消息的工作流程
以下是 Pub-Sub 消息的逐步工作流程 -
- 生产者定期向主题发送消息。
- Kafka 代理存储为该特定主题配置的分区中的所有消息。 它确保消息在分区之间平等共享。 如果生产者发送两个消息并且有两个分区,Kafka 将在第一分区中存储一个消息,在第二分区中存储第二消息。
- 消费者订阅特定主题。
- 一旦消费者订阅主题,Kafka 将向消费者提供主题的当前偏移,并且还将偏移保存在 Zookeeper 系统中。
- 消费者将定期请求 Kafka (如100 Ms)新消息。
- 一旦 Kafka 收到来自生产者的消息,它将这些消息转发给消费者。
- 消费者将收到消息并进行处理。
- 一旦消息被处理,消费者将向 Kafka 代理发送确认。
- 一旦 Kafka 收到确认,它将偏移更改为新值,并在 Zookeeper 中更新它。 由于偏移在 Zookeeper 中维护,消费者可以正确地读取下一封邮件,即使在服务器暴力期间。
- 以上流程将重复,直到消费者停止请求。
- 消费者可以随时回退/跳到所需的主题偏移量,并阅读所有后续消息。
# 队列消息/用户组的工作流
在队列消息传递系统而不是单个消费者中,具有相同组 ID 的一组消费者将订阅主题。 简单来说,订阅具有相同 Group ID 的主题的消费者被认为是单个组,并且消息在它们之间共享。 让我们检查这个系统的实际工作流程。
- 生产者以固定间隔向某个主题发送消息。
- Kafka存储在为该特定主题配置的分区中的所有消息,类似于前面的方案。
- 单个消费者订阅特定主题,假设 Topic-01 为 Group ID 为 Group-1 。
- Kafka 以与发布 - 订阅消息相同的方式与消费者交互,直到新消费者以相同的组 ID 订阅相同主题Topic-01 1 。
- 一旦新消费者到达,Kafka 将其操作切换到共享模式,并在两个消费者之间共享数据。 此共享将继续,直到用户数达到为该特定主题配置的分区数。
- 一旦消费者的数量超过分区的数量,新消费者将不会接收任何进一步的消息,直到现有消费者取消订阅任何一个消费者。 出现这种情况是因为 Kafka 中的每个消费者将被分配至少一个分区,并且一旦所有分区被分配给现有消费者,新消费者将必须等待。
- 此功能也称为使用者组。 同样,Kafka 将以非常简单和高效的方式提供两个系统中最好的。
# kafka安装
直接到官网下载即可 Apache Kafka (opens new window)
然后我们修改一下日志地址(自己按照自己的需求去改)
1.打开config目录下的server.properties文件
2.修改log.dirs=E:\\kafkalogs
3.打开config目录下的zookeeper.properties文件
4.修改dataDir=E:\\kafka\\zookeeper
2
3
4
修改完毕后我们就可以开始启动zookeeper和kafka了,首先是zookeeper,我们以管理员的身份启动cmd,切换到kafka的跟路径,然后执行下面的命令启动zookeeper
.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
出现下面这个说明安装成功
然后启动kafka,同样是切换到这个目录,使用下面的命令启动
.\bin\windows\kafka-server-start.bat .\config\server.properties
# 命令行使用
# 创建主题
下面我们来介绍一下命令行如何操作,首先我们创建一个主题 xiaoyou
.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic xiaoyou
# 创建一个生产者生产消息
.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic xiaoyou
# 创建一个消费者来接收消息
.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic xiaoyou --from-beginning
# 测试
我们使用生产者来发送消息,然后消费者这边就可以接收到消息了
# Go语言操作
先下载依赖
go get github.com/Shopify/sarama
# 生产者代码
我们使用sarama来向kafka发送消息
package main
import (
"fmt"
"github.com/Shopify/sarama"
)
// 基于sarama第三方库开发的kafka client
func main() {
// 初始配置
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll // 发送完数据需要leader和follow都确认
config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
config.Producer.Return.Successes = true // 成功交付的消息将在success channel返回
// 构造一个消息
msg := &sarama.ProducerMessage{}
msg.Topic = "xiaoyou"
msg.Value = sarama.StringEncoder("hello,Word")
// 连接kafka
client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
if err != nil {
fmt.Println("producer closed, err:", err)
return
}
defer client.Close()
// 发送消息
pid, offset, err := client.SendMessage(msg)
if err != nil {
fmt.Println("send msg failed, err:", err)
return
}
fmt.Printf("pid:%v offset:%v\n", pid, offset)
}
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
消费者代码如下
package main
import (
"fmt"
"github.com/Shopify/sarama"
)
// kafka consumer
func main() {
// 连接 kafka
consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
if err != nil {
fmt.Printf("fail to start consumer, err:%v\n", err)
return
}
// 根据topic取到所有的分区
partitionList, err := consumer.Partitions("xiaoyou")
if err != nil {
fmt.Printf("fail to get list of partition:err%v\n", err)
return
}
// 遍历所有的分区
for partition := range partitionList {
// 针对每个分区创建一个对应的分区消费者
pc, err := consumer.ConsumePartition("xiaoyou", int32(partition), sarama.OffsetNewest)
if err != nil {
fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
return
}
defer pc.AsyncClose()
// 从每个分区消费信息
func(sarama.PartitionConsumer) {
for msg := range pc.Messages() {
fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key,string(msg.Value))
}
}(pc)
}
}
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
参考: