sarama 源码学习

本文主要是介绍sarama 源码学习，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

ConsumerGroup

Kafka 消费的过程似乎大致上可以分为 ConsumerGroup 的 Rebalance 协议和 Partition 消费协议两部分，这里先看 ConsumerGroup 部分。

ConsumerGroup 的 Rebalance 是 Kafka 消费侧扩展性的体现，通过将 Partition 分配给 Consumer 实现扩展消费能力，在增加 Consumer 时执行 Rebalance，将 Partition 重新均摊给新加入的 Consumer，是一个 pre-shard 的模型。

Rebalance 流程

每个 ConsumerGroup 有五种状态：

• Empty：无任何活跃 Consumer 存在；
• Stable：已完成 Rebalance，可供稳定消费；
• PreparingRebalance：情况发生变化，有新成员加入或旧成员心跳丢失，需要重新 Balance，要求所有成员重新加入 Group；
• CompletingRebalance：所有成员均已入组，各成员等待分配计划；
• Dead：Group 生命周期结束，可能因为 session 过期，或者 Group 迁移到其他 Group Coordinator；

这篇文章中的图很不错，借一下侵删：

ConsumerGroup 的这几个状态大致上都是服务于 Rebalance 流程的：

1. 一个新的 Consumer 想要加入 ConsumerGroup，发起 JoinGroup 请求；
2. Kafka 将 ConsumerGroup 设置为 PreparingRebalancing 状态，递增 ConsumerGroup 的 generation 值，通过心跳的响应通知现所有成员退出，等待它们重新加入；
3. 选举其中一位成员成为 ConsumerGroup 的 leader；
4. 为所有的成员发送 Join 成功的元信息；
5. Follower 向 Coordinator 发送同步请求申请同步 Partition 分配关系；
6. Leader 按 Balance 策略生成 Partition 的分配关系，向 Coordinator 发送 SyncGroup 请求发送分配关系；
7. Coordinator 回复所有成员各自的 Partition 分配关系；

二阶段 Balance

为什么还要在这里进行一次 Group Leader 的选举呢？

《Kafka Client-side Assignment Proposal》这篇讲到过去的 Kafka 直接通过 Coordinator 服务端来直接管理 Partition 与 Consumer 的分配，客户端向 Coordinator 发送 JoinGroup 请求即可拿到自己分配得到的 Partition 列表。这一机制足够简单，但有一些场景会对分配策略有特殊的需求，需要自定义分配策略时服务端分配便不够灵活，比如：

• Co-partitioning：比如对两个 Topic 做 join，需要将两个 Topic 按相映射的 Partition 来分配给同一个 Consumer；
• Sticky Partitioning：对于有状态的 Consumer，希望重启后仍能恢复原有的 Partition 关系而不要 rebalance；
• Redundant partitioning：比如搜索引擎需要建多份冗余的索引，希望能使单个 Partition 被多个 Consumer 所消费；
• Metadata-based assignment：比如让 Consumer Group 做到 rack aware，消费来自本机架的 Partition；

为了支持这类高级的分配策略甚至未知的分配策略，Kafka 将分配策略下放给了客户端。使新版的协议增加了一个 Group Leader 的选举环节，由作为 Group Leader 的提供 Balance 策略同步给 Coordinator，再使 Coordinator 仲裁过的 Claim 列表同步给其他客户端。

分配 Partition 和同步分配结果两个阶段需所有成员参与协调，对应 PreparingRebalance 和 CompletingBalance 两个状态。

RPC

与 Rebalance 相关有 JoinGroup 和 SyncGroup 两个接口，再加上 Heartbeat 接口。

JoinGroup 接口演进到了第六版：

JoinGroup Request (Version: 6) => group_id session_timeout_ms rebalance_timeout_ms member_id group_instance_id protocol_type [protocols] TAG_BUFFER group_id => COMPACT_STRINGsession_timeout_ms => INT32rebalance_timeout_ms => INT32member_id => COMPACT_STRINGgroup_instance_id => COMPACT_NULLABLE_STRINGprotocol_type => COMPACT_STRINGprotocols => name metadata TAG_BUFFER name => COMPACT_STRINGmetadata => COMPACT_BYTESJoinGroup Response (Version: 6) => throttle_time_ms error_code generation_id protocol_name leader member_id [members] TAG_BUFFER throttle_time_ms => INT32error_code => INT16generation_id => INT32protocol_name => COMPACT_STRINGleader => COMPACT_STRINGmember_id => COMPACT_STRINGmembers => member_id group_instance_id metadata TAG_BUFFER member_id => COMPACT_STRINGgroup_instance_id => COMPACT_NULLABLE_STRINGmetadata => COMPACT_BYTES

当有任何新的 Consumer 发起 JoinGroup 后，Coordinator 会进入 PreparingBalance 状态，递增 generationID，随后等待活跃的所有 Consumer 重新加入 Group，等待的期限为 rebalance_timeout_ms（默认值为 60s）。

里面有两个 timeout 值得注意：

• session_timeout_ms：表示 Coordinator 如果超过该超时值没有收到心跳，则认为 session 过期；
• rebalance_timeout_ms：表示 Coordinator 等待所有 Consumer 重新申请加入的最大时限；

假如我扩容 100 个 Consumer，要等多久生效？显然不能是 60s。那么 Coordinator 怎么知道要等到什么时候呢？等待所有活跃 session 的 Consumer 都发送过 JoinGroup 便可以了。每个 Consumer 每 3s 与 Coordinator 发一次心跳，这也意味着一次 rebalance 大约需要 3s 左右的等待。如果减少心跳的时间间隔，Rebalance 的生效时间应能够相应减少。

也就是说，所有客户端的 JoinGroup 会阻塞直到所有活跃 Session 的 Consumer 皆执行了 JoinGroup 为止。随后 ConsumerGroup 将进入 CompletingBalance 状态。

SyncGroup 演进到了 Version 4：

SyncGroup Request (Version: 4) => group_id generation_id member_id group_instance_id [assignments] TAG_BUFFER group_id => COMPACT_STRINGgeneration_id => INT32member_id => COMPACT_STRINGgroup_instance_id => COMPACT_NULLABLE_STRINGassignments => member_id assignment TAG_BUFFER member_id => COMPACT_STRINGassignment => COMPACT_BYTESSyncGroup Response (Version: 4) => throttle_time_ms error_code assignment TAG_BUFFER throttle_time_ms => INT32error_code => INT16assignment => COMPACT_BYTES

客户端通过 JoinGroup 响应中的 leader 字段检查自己是否当选为 Leader。如果是 Leader，则执行分配策略，将分配的结果 assignments 通过 SyncGroup 发送出去。若非 Leader，也同样调用 SyncGroup，但是不携带 assignments 字段。

SyncGroup 的阻塞等待也与 JoinGroup 类似，当所有 Consumer 皆发起 SyncGroup 之前，所有请求方将一直阻塞等待 SyncGroup 返回。SyncGroup 返回也意味着 ConsumerGroup 进入 Stable 状态。返回的响应体中携带有每个 Consumer 最终得到的 Partition 分配结果。

Generation 值

注意到前面 JoinGroup 会返回 generation 值，而在 JoinGroup 时需携带着这个 generation 值。这个 generation 大致上与 “epoch” 的含义类似，每代 generation 对应一代选举得出的 Leader。似乎但凡 Leader 选举的场景都需要对付这一问题，便是 Leader 自身若因为 Full GC 或长阻塞被摘除了 Leader，但它自身仍以为自己是 Leader，仍会凭据 Leader 身份去找其他节点做管理操作。

递增的 generation 值或者 epoch 值可以用于避免旧 Leader 做出破坏性的操作，做到：

1. 在每次选举后，将递增的 generation 值为每个节点加入共识;
2. 每次请求携带 generation 值；

这一来每个节点收到请求时，皆判断一下请求的 generation 值是否小于自己从共识中得到的 generation 值，如果小于则返回错误码，令对方重试。

时序关系

画一个时序图总结一下两阶段 Rebalance 的流程：

sarama ConsumerGroup 的基本使用

sarama 的 ConsumerGroup 的使用上，参考仓库中的例子 https://github.com/Shopify/sarama/blob/master/examples/consumergroup/main.go，大约是：

1. NewConsumerGroup(addrs []string, groupID string, config *Config) 创建 ConsumerGroup 对象；
2. 调用 ConsumerGroup 接口的 Consume(ctx context.Context, topics []string, handler ConsumerGroupHandler) error 方法，订阅 topic，递 ConsumerGroupHandler 对象，需要注意每当遇到 Rebalance，Consume() 函数便会退出，调用方需要在一个无限循环中不停地调用 Consume()；
3. 在 ConsumerGroupHandler 对象提供 ConsumeClaim(ConsumerGroupSession, ConsumerGroupClaim) error 方法，通过 ConsumerGroupClaim 对象的 Messages() 得到的 channel 消费一个 Partition 的消息；

其中每个 partition 与 consumer 的分配关系称作一个 “claim”。

sarama 内部管理着 ConsumerGroupClaim 的生命周期，大约是：

1. 首先执行 ConsumerGroupClaim 接口的 Setup() 回调；
2. 对每个 claim 单独隐式在一个 goroutine 里跑 ConsumeClaim() 中的 for m := range claim.Messages() 循环；
3. 当所有 ConsumeClaim 的 goroutine 退出之后，再走到 Cleanup()；

一组 ConsumerGroupClain 这一轮的生命周期称作一个 session。session 的退出发生在 ctx 退出，或者 partition rebalance。session 要求客户端与 coordinator 保持一定的心跳，原版 kafka 客户端为此有一条 session.timeout.ms 的配置，客户端需要在时间范围内对 coordinator 发送心跳，不然将视为该客户端退出而出发 Rebalance。

Consume

// Consume implements ConsumerGroup.
func (c *consumerGroup) Consume(ctx context.Context, topics []string, handler ConsumerGroupHandler) error {// Ensure group is not closedselect {case <-c.closed:return ErrClosedConsumerGroupdefault:}c.lock.Lock()defer c.lock.Unlock()// Quick exit when no topics are providedif len(topics) == 0 {return fmt.Errorf("no topics provided")}// Refresh metadata for requested topicsif err := c.client.RefreshMetadata(topics...); err != nil {return err}// Init sessionsess, err := c.newSession(ctx, topics, handler, c.config.Consumer.Group.Rebalance.Retry.Max)if err == ErrClosedClient {return ErrClosedConsumerGroup} else if err != nil {return err}// loop check topic partition numbers changed// will trigger rebalance when any topic partitions number had changed// avoid Consume function called again that will generate more than loopCheckPartitionNumbers coroutinego c.loopCheckPartitionNumbers(topics, sess)// Wait for session exit signal<-sess.ctx.Done()// Gracefully release session claimsreturn sess.release(true)
}

大致上做的这几件事情：

1. 通过 c.closed 检查 ConsumerGroup 是否已关闭；
2. 根据 topics 抓取 metadata：c.client.RefreshMetadata(topics…)；
3. 初始化 session；
4. 启动一个 goroutine 来检查 partition 的变化；
5. 等待 session 退出；
6. 释放 session；

Session 创建流程

func (c *consumerGroup) newSession(ctx context.Context, topics []string, handler ConsumerGroupHandler, retries int) (*consumerGroupSession, error) {coordinator, err := c.client.Coordinator(c.groupID)if err != nil {if retries <= 0 {return nil, err}return c.retryNewSession(ctx, topics, handler, retries, true)}// Join consumer groupjoin, err := c.joinGroupRequest(coordinator, topics)if err != nil {_ = coordinator.Close()return nil, err}switch join.Err {case ErrNoError:c.memberID = join.MemberIdcase ErrUnknownMemberId, ErrIllegalGeneration: // reset member ID and retry immediatelyc.memberID = ""return c.newSession(ctx, topics, handler, retries)case ErrNotCoordinatorForConsumer: // retry after backoff with coordinator refreshif retries <= 0 {return nil, join.Err}return c.retryNewSession(ctx, topics, handler, retries, true)case ErrRebalanceInProgress: // retry after backoffif retries <= 0 {return nil, join.Err}return c.retryNewSession(ctx, topics, handler, retries, false)default:return nil, join.Err}// Prepare distribution plan if we joined as the leadervar plan BalanceStrategyPlanif join.LeaderId == join.MemberId {members, err := join.GetMembers()if err != nil {return nil, err}plan, err = c.balance(members)if err != nil {return nil, err}}// Sync consumer groupsync, err := c.syncGroupRequest(coordinator, plan, join.GenerationId)if err != nil {_ = coordinator.Close()return nil, err}switch sync.Err {case ErrNoError:case ErrUnknownMemberId, ErrIllegalGeneration: // reset member ID and retry immediatelyc.memberID = ""return c.newSession(ctx, topics, handler, retries)case ErrNotCoordinatorForConsumer: // retry after backoff with coordinator refreshif retries <= 0 {return nil, sync.Err}return c.retryNewSession(ctx, topics, handler, retries, true)case ErrRebalanceInProgress: // retry after backoffif retries <= 0 {return nil, sync.Err}return c.retryNewSession(ctx, topics, handler, retries, false)default:return nil, sync.Err}// Retrieve and sort claimsvar claims map[string][]int32if len(sync.MemberAssignment) > 0 {members, err := sync.GetMemberAssignment()if err != nil {return nil, err}claims = members.Topicsc.userData = members.UserDatafor _, partitions := range claims {sort.Sort(int32Slice(partitions))}}return newConsumerGroupSession(ctx, c, claims, join.MemberId, join.GenerationId, handler)
}

Rebalance 相关流程大部分位于这里，newSession 内的流程大致上是：

1. 首先通过 c.groupID 找到 coordinator 对象，如果失败则重试；
2. 调用 c.joinGroupRequest() 对 coordinator 发送 joinGroupRequest；如果 join 成功，可以得到一个唯一的 memberID；
   • 如果 ErrUnknownMemberId, ErrIllegalGeneration，则立即重试；
   • 如果 ErrUnknownMemberId, ErrIllegalGeneration，则立即重试；
   • 如果 Coordinator 过期、Consumer Group 重平衡，则 backoff 后重试；
3. 如果该 ConsumerGroup 客户端被选为 Group 内的 leader，则由它调用 c.balance() 按照 BalanceStrategyPlan 进行 partition 分配，产出 BalanceStrategyPlan 类型的 plan；
4. 调用 c.syncGroupRequest(coordinator, plan, join.GenerationId)，得到 SyncGroupResponse；
   • 如果是 group leader，则 plan 参数中携带着自身产出的分配计划；如果非 group leader，则 plan 为空；
   • 这里的错误处理与第二步的 joinGroupRequest 类似，大致上也是重试；
5. 根据 SyncGroupResponse 中的 claims 列表，得出自己的 claims 表，即每个 topic 对应的 partitionID 的列表；
6. 最后根据 claims 列表得到 consumerGroupSession；

然后看一下 newConsumerGroupSession 相关的流程：

type consumerGroupSession struct {parent       *consumerGroupmemberID     stringgenerationID int32handler      ConsumerGroupHandlerclaims  map[string][]int32offsets *offsetManagerctx     context.Contextcancel  func()waitGroup       sync.WaitGroupreleaseOnce     sync.OncehbDying, hbDead chan none
}func newConsumerGroupSession(ctx context.Context, parent *consumerGroup, claims map[string][]int32, memberID string, generationID int32, handler ConsumerGroupHandler) (*consumerGroupSession, error) {// init offset manageroffsets, err := newOffsetManagerFromClient(parent.groupID, memberID, generationID, parent.client)if err != nil {return nil, err}// init contextctx, cancel := context.WithCancel(ctx)// init sessionsess := &consumerGroupSession{parent:       parent,memberID:     memberID,generationID: generationID,handler:      handler,offsets:      offsets,claims:       claims,ctx:          ctx,cancel:       cancel,hbDying:      make(chan none),hbDead:       make(chan none),}// start heartbeat loopgo sess.heartbeatLoop()// create a POM for each claimfor topic, partitions := range claims {for _, partition := range partitions {pom, err := offsets.ManagePartition(topic, partition)if err != nil {_ = sess.release(false)return nil, err}// handle POM errorsgo func(topic string, partition int32) {for err := range pom.Errors() {sess.parent.handleError(err, topic, partition)}}(topic, partition)}}// perform setupif err := handler.Setup(sess); err != nil {_ = sess.release(true)return nil, err}// start consumingfor topic, partitions := range claims {for _, partition := range partitions {sess.waitGroup.Add(1)go func(topic string, partition int32) {defer sess.waitGroup.Done()// cancel the as session as soon as the first// goroutine exitsdefer sess.cancel()// consume a single topic/partition, blockingsess.consume(topic, partition)}(topic, partition)}}return sess, nil
}

流程上大致上：

1. 首先初始化 offsetManager；
2. 派生新的 ctx，由该 ctx 控制 Session 的退出生命周期；
3. 启动 heartbeatLoop()；
4. 为每个 Partition 创建 partitionOffsetManager，简称 POM；
5. 调用 Setup(sess) 执行 Handler 的 Setup 逻辑；
6. 对每个 claim 中的每个 Partition 遍历，执行 sess.consume(topic, partition)；
7. 任意一个 Partition 退出消费，则关闭整个 sess 的 ctx；

Session 的生命周期与一次 Consume() 是一致的，Session 若因为任何原因而退出，则应由调用方去重新 Consume()。

消费

func (s *consumerGroupSession) consume(topic string, partition int32) {// quick exit if rebalance is dueselect {case <-s.ctx.Done():returncase <-s.parent.closed:returndefault:}// get next offsetoffset := s.parent.config.Consumer.Offsets.Initialif pom := s.offsets.findPOM(topic, partition); pom != nil {offset, _ = pom.NextOffset()}// create new claimclaim, err := newConsumerGroupClaim(s, topic, partition, offset)if err != nil {s.parent.handleError(err, topic, partition)return}**// handle errorsgo func() {for err := range claim.Errors() {s.parent.handleError(err, topic, partition)}}()// trigger close when session is donego func() {select {case <-s.ctx.Done():case <-s.parent.closed:}claim.AsyncClose()}()// start processingif err := s.handler.ConsumeClaim(s, claim); err != nil {s.parent.handleError(err, topic, partition)}// ensure consumer is closed & drainedclaim.AsyncClose()for _, err := range claim.waitClosed() {s.parent.handleError(err, topic, partition)}
}

流程大致上是：

1. 检查 Session 或者 ConsumerGroup 是否退出；
2. 从 OffsetManager 中取得恢复消费的 Offset；
3. 通过 topic、partition 和 offset 调用 newConsumerGroupClaim 创建 Claim 对象，在里面调用 Consumer 对象的 ConsumePartition 方法执行真正的消费；
4. 开 goroutine 消费 claim.Errors() 将报错交给 s.parent.handleError；
5. 开 goroutine 侦听 ctx 退出，则关闭 claim；
6. 回调 handler 的 ConsumeClaim 方法供真正消费，阻塞直到 Session 退出为止；
7. 执行退出并等待退出；

心跳

心跳的行为似乎比较简单，发送 heartbeatRequest，一旦心跳循环因为任何原因退出，该 session 的 ctx 便宣告结束。

type HeartbeatRequest struct {GroupId      stringGenerationId int32MemberId     string
}type HeartbeatResponse struct {Err KError
}

如果 Coordinator 进入 PrepareRebalance 状态，所有的 Consumer 可以通过心跳响应的 Err 字段得到通知，而重新发起 JoinGroup。

总结

• ConsumerGroup 扮演着最上层的用户接口，Rebalance 的生命周期管理是它的主要内容。
• ConsumerGroup 相关的 RPC 接口就是 JoinRequest、SyncGroup 和 Heartbeat 这三个。
• 过去 Kafka 是 Coordinator 服务端直接分配 Partition，后来期望更灵活的 Partition 分配，为此实现的二阶段 Rebalance 协议。
• ConsumerGroup 在各种意外情况的处理方面似乎都走的比较稳重的办法，就是协调重新走一遍整个流程。
• 每一轮 Rebalance 的生命周期对应一个 Session，每当新成员加入或者退出，则所有成员进入新的 Session。
• 具体的 Partition 消费与位移管理部分似乎在 consumer.go 中更多，后面详细看好了。

---

Consumer

这里过一下消费的流程。
consumer 的结构体很简单，里面主要是 partitionConsumer 和 brokerConsumer 两个表：

type consumer struct {conf            *Configchildren        map[string]map[int32]*partitionConsumerbrokerConsumers map[*Broker]*brokerConsumerclient          Clientlock            sync.Mutex
}

从上篇笔记中记得 ConsumePartition() 是消费的入口，看代码大约是初始化一个 partitionConsumer，并启动 partitionConsumer 的 dispatcher 和 responseFeeder 两个 goroutine，将 brokerConsumer 的引用交给 partitionConsumer。

注释中提到除非调用 AsyncClose() 或者 Close() 方法，ConsumePartition() 启动的消费过程会永不停止，遇到错误，就不停重试，如果开启 Consumer.Return.Errors，则可以将收到的错误交给 Errors 这个 channel 允许用户处理。

brokerConsumer 是什么呢？当 Consumer 需要从 Broker 订阅多个 Topic 时，会使用单独的一个连接来消费数据，再将数据按 partition 分给不同的 partitionConsumer。Consumer、partitionConsumer、brokerConsumer 乃至 broker 之间大致上似乎是这样的关系：

整理一下 partitionConsumer 与 brokerConsumer 之间的交互还是有点复杂的，大概是：

1. partitionConsumer 通过 broker.input 这个 chan 加入 brokerConsumer 的订阅；
2. 加入订阅后，brokerConsumer 每次 fetchNewMessages 得到一批消息，会发给每个 partitionConsumer 的 feeder chan；
3. 每个 partitionConsumer 解析 feeder chan 的数据得到 对应的消息列表，同时设置自身的 responseResult，设置 brokers.acks 这个 WaitGoup 执行 Done()，表示处理完毕该批次；
4. brokerConsumer 等待每个 partitonConsumer 处理完毕该批次，处理所有 partitionConsumer 的 responseResult 后循环 fetchNewMessages；

这里的交互还是比较 go 特色的，直觉上 java 原版客户端该不会做这么麻烦的交互。

brokerConsumer 的 input chan 有点像服务端的 accept，会有多个 partitionConsumer 动态加入、解除对 brokerConsumer 的订阅。

brokerConsumer 会负责发起 FetchRequest 的主循环，每轮迭代拉取一批消息，发送给每个 partitionConsumer 再转给用户的 messages chan。通过 acks 这个 WaitGroup 来协调每个 partitionConsumer 对这一批次的消息的处理节奏。

如果用户消费 messages chan 有超时，parseResponse 会返回 errTimeout，brokerConsumer 会依据 errTimeout 而暂停 fetchRequest。

partitionConsumer

partitonConsumer 会启动 dispatcher 和 responseFeeder 两个 goroutine，其中 dispatcher goroutine 用于跟踪 broker 的变化，偏元信息性质的控制侧，而 responseFeeder 用于跟踪消息的到来，偏数据侧。

func (child *partitionConsumer) dispatcher() {for range child.trigger {select {case <-child.dying:close(child.trigger)case <-time.After(child.computeBackoff()):if child.broker != nil {child.consumer.unrefBrokerConsumer(child.broker)child.broker = nil}Logger.Printf("consumer/%s/%d finding new broker\n", child.topic, child.partition)if err := child.dispatch(); err != nil {child.sendError(err)child.trigger <- none{}}}}if child.broker != nil {child.consumer.unrefBrokerConsumer(child.broker)}child.consumer.removeChild(child)close(child.feeder)
}// ...func (child *partitionConsumer) dispatch() error {if err := child.consumer.client.RefreshMetadata(child.topic); err != nil {return err}var leader *Brokervar err errorif leader, err = child.consumer.client.Leader(child.topic, child.partition); err != nil {return err}child.broker = child.consumer.refBrokerConsumer(leader)child.broker.input <- childreturn nil}

dispatcher 这个 goroutine 用于发现 broker 的变化。它会侦听 dispatcher.trigger 这个 channel 的通知，来发现 Partition 的 Leader 变化。而 trigger 这个 channel 的更新来自 brokerConsumer 对象。

最后 child.broker.input child 这一句，相当于使 partitionConsumer 加入 brokerConsumer 的订阅。

不过这里不是很明白 dispatcher 这个 goroutine 里为啥没有对 partitionConsumer 对象上锁。

func (child *partitionConsumer) responseFeeder() {var msgs []*ConsumerMessageexpiryTicker := time.NewTicker(child.conf.Consumer.MaxProcessingTime)firstAttempt := truefeederLoop:for response := range child.feeder {msgs, child.responseResult = child.parseResponse(response)if child.responseResult == nil {atomic.StoreInt32(&child.retries, 0)}for i, msg := range msgs {messageSelect:select {case <-child.dying:child.broker.acks.Done()continue feederLoopcase child.messages <- msg:firstAttempt = truecase <-expiryTicker.C:if !firstAttempt {child.responseResult = errTimedOutchild.broker.acks.Done()remainingLoop:for _, msg = range msgs[i:] {select {case child.messages <- msg:case <-child.dying:break remainingLoop}}child.broker.input <- childcontinue feederLoop} else {// current message has not been sent, return to select// statementfirstAttempt = falsegoto messageSelect}}}child.broker.acks.Done()}expiryTicker.Stop()close(child.messages)close(child.errors)
}

child.feed 这个 channel 也是来自 brokerConsumer。大约是处理来自 brokerConsumer 的消息，转发给 messages chan。

值得留意有一个配置项目 child.conf.Consumer.MaxProcessingTime，默认值为 100ms，看注释它的意思是如果朝 messages chan 写入超过 100ms 仍未成功，则停止再向 Broker 发送 fetch 请求。这一流程的实现有一些细节：

1. 通过 firstAttempt 变量判断是否第一次超时，只有当第二次超时才跑暂停 Broker 发送 fetch 请求的流程
2. 设置 child.responseResult 为 errTimeout，以通知 brokerConsumer 暂停。
3. 将当前批次的消息列表消化掉，如果中途 partitionConsumer 退出，则停止消化当前批次的消息；
4. child.broker.input child，将 partitionConsumer 重新加入 brokerConsumer 的订阅；

brokerConsumer

brokerConsumer 的结构体如下：

type brokerConsumer struct {consumer         *consumerbroker           *Brokerinput            chan *partitionConsumernewSubscriptions chan []*partitionConsumersubscriptions    map[*partitionConsumer]nonewait             chan noneacks             sync.WaitGrouprefs             int}

初始化 brokerConsumer 时会产生两个 goroutine：

1. bc.subscriptionManager：侦听 input chan，获取加入订阅的 partitionConsumer；
2. bc.subscriptionConsumer：循环发起 FetchRequest，向 broker 拉取一批消息，协调发给 partitionConsumer；

subscriptionConsumer 相当于 FetchRequest 的主循环流程：

//subscriptionConsumer ensures we will get nil right away if no new subscriptions is availablefunc (bc *brokerConsumer) subscriptionConsumer() {<-bc.wait // wait for our first piece of workfor newSubscriptions := range bc.newSubscriptions {bc.updateSubscriptions(newSubscriptions)if len(bc.subscriptions) == 0 {// We're about to be shut down or we're about to receive more subscriptions.// Either way, the signal just hasn't propagated to our goroutine yet.<-bc.waitcontinue}response, err := bc.fetchNewMessages()if err != nil {Logger.Printf("consumer/broker/%d disconnecting due to error processing FetchRequest: %s\n", bc.broker.ID(), err)bc.abort(err)return}bc.acks.Add(len(bc.subscriptions))for child := range bc.subscriptions {child.feeder <- response}bc.acks.Wait()bc.handleResponses()}}

而 subscriptionManager 用于管理订阅的加入，并通过 bc.newSubscriptions 来通知给 subscriptionConsumer。

// The subscriptionManager constantly accepts new subscriptions on `input` (even when the main subscriptionConsumer// goroutine is in the middle of a network request) and batches it up. The main worker goroutine picks// up a batch of new subscriptions between every network request by reading from `newSubscriptions`, so we give// it nil if no new subscriptions are available. We also write to `wait` only when new subscriptions is available,// so the main goroutine can block waiting for work if it has none.func (bc *brokerConsumer) subscriptionManager() {var buffer []*partitionConsumerfor {if len(buffer) > 0 {select {case event, ok := <-bc.input:if !ok {goto done}buffer = append(buffer, event)case bc.newSubscriptions <- buffer:buffer = nilcase bc.wait <- none{}:}} else {select {case event, ok := <-bc.input:if !ok {goto done}buffer = append(buffer, event)case bc.newSubscriptions <- nil:}}}done:close(bc.wait)if len(buffer) > 0 {bc.newSubscriptions <- buffer}close(bc.newSubscriptions)}

总结

• consumer 与 broker 算是个多对多的关系，如果有多个 partition 位于一个 broker，那么通过单个 brokerConsumer 与之统一交互。
• 因此 partitionConsumer 与 brokerConsumer 属于一个订阅的关系，partitonConsumer 关注的点是将自己加入订阅并处理订阅的内容，由 brokerConsumer 驱动 FetchRequest 循环；
• brokerConsumer 使用一个 WaitGroup 来协调多个 partitionConsumer 的执行节奏与结果。
• 里面这些 goroutine 好像都没怎么上锁，不是很明白会不会有线程安全问题。

这篇关于sarama 源码学习的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---