项目用于记录我学习的kafka消息队列的过程，学习笔记和完成的源码都在github中提供，GitHub链接在文章的最后，因为文章过长，会分多个博客记录，此次项目是学习q1mi博客视频所做，将用到，kafka消息队列、etcd分布式管理器、context上下文、logrus程序进程记录日志、tail文件读取包、其中还有ES和Kibana。
kafka执行流程示意图 kafka保证消息安全的三种确认模式ACK 0 1 all0：生产者只负责生产不需要恢复是否保存 速度快 安全性低
1：只需要leader回复ack确认
all:全部需要回复ack
kafka自带管理机zookeeper主要用到功能 1、zookeeper 自动注册发现 用在分布式动态扩容；2、分布式锁
kafka的操作步骤 首先生产着的配置config//生产者配置 config:=sarama.NewConfig() //配置kafka的ack回传级别 config.Producer.RequiredAcks=sarama.WaitForAll //发送到那个分区 config.Producer.Partitioner=sarama.NewRandomPartitioner // 成功交付信息 config.Producer.Return.Successes=true连接kafka
// 连接kafka
client,err:=sarama.NewSyncProducer([]string{"127.0.0.1:9092"},config)
if err!=nil {
fmt.Println("producer closer err:",err)
return
}
defer client.Close()
创建消息
// 构造一个消息
msg:=&sarama.ProducerMessage{}
msg.Topic="shopping"
msg.Value=sarama.StringEncoder("2022kafka")
发送消息
// 发送消息
pid,offset,err:=client.SendMessage(msg)
if err!=nil {
fmt.Println("client send err :",err)
return
}
fmt.Println(pid,offset)
注：下个博客将开始搭建日志收集项目
GitHub链接：https://github.com/Libing0804/kafka_log_collect
