Kafka

哪些场景适合消息队列？

1. 异步处理
   • 场景：用户注册，客户端发送消息给server，server发送内容给db，并且发送消息给外部服务（邮件验证、短信验证）。外部服务处理后返回消息给server，server返回内容给客户端。
   • 外部服务不是实时性很高的内容，也就是不必要的流程，不具备紧急性，所以在server和外部服务之间，加入消息队列。
   • 最终的流程变为，只要db写入完成，注册流程就完成了，而不需要等待外部服务。
2. 系统解耦
   • 微服务之间通信：restful、socket、rpc方案
   • rpc方案的缺点：
     1. 远程调用需要知道函数名、参数列表、参数含义
     2. 也就是微服务之间的耦合度很高，微服务必须知道对方的实现
   • 消息队列解耦：
     • 只使用消息进行通信，不关心对方具体的实现和如何处理；
3. 流量削峰
   • 场景：请求量大于我们系统的处理能力
   • 在server和db间加入消息队列，和多个消费者，这样请求就会在消息队列中排队，经由多个消费者对排队的内容进行db处理
4. 日志处理
   • 场景：多个功能模块，实现一个功能，日志打印，日志入库（用来分析统计）
   • 日志入库功能，我们发现是一个不紧急的功能。所以使用消息队列，排队后，再写入日志。
   • 写消息队列是顺序写，而db是磁盘写，写消息队列要快很多。

• 总结（当我们看是否需要消息队列时）：
  1. 梳理流程
  2. 是否必要，是否紧急
  3. 非必要不紧急的人物可以使用消息队列处理

生产消费模型

• 消费者：
  1. 推模式（消费者点不确定、不稳定，推模式不好）
  2. 拉模式

1. 点对点
2. 订阅发布

kafka基本原理

• kafka架构图
  

1. zk + kafka集群构成
   • zk：负责注册、存储、选举
2. broker：kafka节点，一个kafka进程，又多个partition
3. partition：每一个分区都会有副本，即replica
4. 主broker，即controller
5. 主partition，即leader，由controller选举出来
6. topic主题：逻辑概念，有多个paritition，分多个p可以提高吞吐量

• 幂等性：在消息队列中避免消息重复
  • kafka如何处理幂等性？
    • 每个生产者都有pid，每个都有seq
    • 如果发现一个消息的pid和seq有重复，那么该消息将被kafka丢弃
    • 每个生产者收到kakfa的ack，那么seq++
• 有多个partition的副本，那么生产者会写入哪个partition？
  • 写主partition，也就是leader，所以生产者会先去zk中查leader是哪个
• 有多个partition，那么生产者会写入哪个partition？
  1. 随机写
  2. hash(key) % partition
  3. 轮询
• 生产者是如何知道写入到partition的哪个位置的？
  • 每个partition都维护了一个offset，生产者写入offset的位置，然后offset会进行偏移
• 两个选举
  1. controller选举，zk实现
  2. partition选举，controller实现
     • 为什么这里不是zk？因为zk和etcd是一样的，是会保持强一致性的，如果partition也要zk来负责选举，zk压力太大了。
• partition
  1. leader 负责数据读写
  2. follower 负责数据备份 未来可以选举leader
  • partition内部是有序的，但是partition之间由于被多个消费者并发读取，全局是无序的。
  • 如果要保证全局有序性，一个主题指定只有一个partition；

kafka读写机制

生产详细流程

1. producer先找zk，知道leader p0 p1 p2 的位置
2. producer的ack设置有三种方式
   1. ack=0 不需要回复，可能会有数据丢失；吞吐量：20w/s
   2. ack=1 不需等待同步，leader所在broker宕机，可能产生数据丢失
   3. ack=-1 需要等待副本同步，再回复producer；吞吐量：10w/s

消费者消费流程

1. 如何消费消息：
   1. 自动提交，消费者取出消息，offset偏移，也就是指向下一个消息，这种情况下假如消费者消费失败，可能会造成消息丢失
   2. 手动提交，消息正确消费后，由消费者提醒partition，进行offset偏移
2. 消费者如何知道从哪儿开始消费？
   • zk的作用：
     1. 存储上次消费的partition；以及之前的offset；注意：之后的版本offset会存在具体的partition中，存放在特定的topic里；
3. 如何避免重复消费？
   • 为什么会出现重复消费问题？
     • 因为消费者取出了消息，然后也消费了消息，将数据存入了db，但是此时消费者宕机了，也就是没有进行offset提交。此时消费者重启后，会出现重复消费的问题。
   • 避免重复消费：
     • 保证消费者消费数据存入db和offset更新是一个原子操作；
     • low level api：即，将offset存储在db当中，假如消费者宕机了，此时db会进行回滚那么offset的更改也会进行回滚。
4. 消费者组和消费者
   • 消费者组对应一个topic
   • 消费者组里有多个消费者
   • 一个partition不能由多个消费者消费
     • 一个partition只对应一个消费者
     • 一个消费者可以对应多个partition
5. 为什么需要消费者组？
   • 因为相同的消息，需要不同的消费方向

kafka go操作

1
2

go get github.com/IBM/sarama # 操作kafka
go get github.com/bsm/sarama-cluster # 操作kafka集群，已经合并到上一个中了，不需要这个

• high level api
  • producer写入策略只需要指定是随机写、hash还是轮询，然后api帮我们完成选择paritition
  • comsumer的offset存储在partition中
• low level api
  • 我们自己指定如果选择partition进行写入，而且ack方式也需要自己实现
  • comsumer的offset可以自己实现存储位置
• /producer/main.go

  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 42

  package main import ( "bufio" "fmt" "os" "github.com/IBM/sarama" ) // high level api func main() { config := sarama.NewConfig() config.Producer.RequiredAcks = sarama.WaitForLocal config.Producer.Partitioner = sarama.NewRoundRobinPartitioner config.Producer.Return.Successes = true prod, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config) if err != nil { panic(err) } defer prod.Close() msg := &sarama.ProducerMessage{ Topic: "test", Partition: int32(0), Key: sarama.StringEncoder("key"), } ireader := bufio.NewReader(os.Stdin) for { data, _, err := ireader.ReadLine() if err != nil { break } msg.Value = sarama.ByteEncoder(data) partition, offset, err := prod.SendMessage(msg) if err != nil { fmt.Println("Send message failed") } fmt.Printf("partition: %d, offset: %d\n", partition, offset) } }

• /consumer/main.go

  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

  package main import ( "fmt" "sync" "github.com/IBM/sarama" ) var ( wg sync.WaitGroup ) func main() { consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil) if err != nil { panic(err) } defer consumer.Close() partitionlist, err := consumer.Partitions("test") if err != nil { panic(err) } for partition := range partitionlist { pt, err := consumer.ConsumePartition("test", int32(partition), sarama.OffsetNewest) if err != nil { panic(err) } defer pt.AsyncClose() wg.Add(1) go func(sarama.PartitionConsumer) { defer wg.Done() for msg := range pt.Messages() { fmt.Printf("%s partition: %d, offset:%d,key:%s,value:%s", msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value) } }(pt) } wg.Wait() }