golang源码分析：sarama kafka client(part II:消费者)

package main


import (
  "fmt"
  "log"
  "sync"


  "github.com/Shopify/sarama"
)


// 消费者练习


func main() {
  // 生成消费者 实例
  consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
  if err != nil {
    log.Print(err)
    return
  }
  // 拿到 对应主题下所有分区
  partitionList, err := consumer.Partitions("test")
  if err != nil {
    log.Println(err)
    return
  }


  var wg sync.WaitGroup
  wg.Add(1)
  // 遍历所有分区
  for partition := range partitionList {
    //消费者 消费 对应主题的 具体 分区 指定 主题 分区 offset  return 对应分区的对象
    pc, err := consumer.ConsumePartition("test", int32(partition), sarama.OffsetNewest)
    if err != nil {
      log.Println(err)
      return
    }


    // 运行完毕记得关闭
    defer pc.AsyncClose()


    // 去出对应的 消息
    // 通过异步 拿到 消息
    go func(sarama.PartitionConsumer) {
      defer wg.Done()
      for msg := range pc.Messages() {
        fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
      }
    }(pc)
  }
  wg.Wait()
}
复制

package main


import (
  "context"
  "fmt"
  "os"
  "os/signal"
  "sync"


  "github.com/Shopify/sarama"
)


type consumerGroupHandler struct {
  name string
}


func (consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
  for msg := range claim.Messages() {
    fmt.Printf("%s Message topic:%q partition:%d offset:%d  value:%s\n", h.name, msg.Topic, msg.Partition, msg.Offset, string(msg.Value))
    // 手动确认消息
    sess.MarkMessage(msg, "")
  }
  return nil
}


func handleErrors(group *sarama.ConsumerGroup, wg *sync.WaitGroup) {
  wg.Done()
  for err := range (*group).Errors() {
    fmt.Println("ERROR", err)
  }
}


func consume(group *sarama.ConsumerGroup, wg *sync.WaitGroup, name string) {
  fmt.Println(name + "start")
  wg.Done()
  ctx := context.Background()
  for {
    topics := []string{"test"}
    handler := consumerGroupHandler{name: name}
    err := (*group).Consume(ctx, topics, handler)
    fmt.Println("consume group end")
    if err != nil {
      panic(err)
    }
  }
}


func main() {
  var wg sync.WaitGroup
  config := sarama.NewConfig()
  config.Consumer.Return.Errors = false
  config.Version = sarama.V0_10_2_0
  client, err := sarama.NewClient([]string{"localhost:9092"}, config)
  defer client.Close()
  if err != nil {
    panic(err)
  }
  group1, err := sarama.NewConsumerGroupFromClient("c1", client)
  if err != nil {
    panic(err)
  }
  group2, err := sarama.NewConsumerGroupFromClient("c2", client)
  if err != nil {
    panic(err)
  }
  group3, err := sarama.NewConsumerGroupFromClient("c3", client)
  if err != nil {
    panic(err)
  }
  defer group1.Close()
  defer group2.Close()
  defer group3.Close()
  wg.Add(3)
  go consume(&group1, &wg, "c1")
  go consume(&group2, &wg, "c2")
  go consume(&group3, &wg, "c3")
  wg.Wait()
  signals := make(chan os.Signal, 1)
  signal.Notify(signals, os.Interrupt)
  select {
  case <-signals:
  }
}
复制

func NewConsumerGroup(addrs []string, groupID string, config *Config) (ConsumerGroup, error) {
    client, err := NewClient(addrs, config)
    c, err := newConsumerGroup(groupID, client)
}
复制

type ConsumerGroup interface {
  Consume(ctx context.Context, topics []string, handler ConsumerGroupHandler) error
  // Errors returns a read channel of errors that occurred during the consumer life-cycle.
  // By default, errors are logged and not returned over this channel.
  // If you want to implement any custom error handling, set your config's
  // Consumer.Return.Errors setting to true, and read from this channel.
  Errors() <-chan error
  // Close stops the ConsumerGroup and detaches any running sessions. It is required to call
  // this function before the object passes out of scope, as it will otherwise leak memory.
  Close() error
}
复制

type consumerGroup struct {
  client Client


  config   *Config
  consumer Consumer
  groupID  string
  memberID string
  errors   chan error


  lock      sync.Mutex
  closed    chan none
  closeOnce sync.Once


  userData []byte
}
复制

func newConsumerGroup(groupID string, client Client) (ConsumerGroup, error) {
  consumer, err := NewConsumerFromClient(client) 
}
复制

func NewConsumerFromClient(client Client) (Consumer, error) {
  cli := &nopCloserClient{client}
  return newConsumer(cli)
}
复制

func newConsumer(client Client) (Consumer, error) {
}
复制

type consumer struct {
  conf            *Config
  children        map[string]map[int32]*partitionConsumer
  brokerConsumers map[*Broker]*brokerConsumer
  client          Client
  lock            sync.Mutex
}
复制

func (c *consumerGroup) Consume(ctx context.Context, topics []string, handler ConsumerGroupHandler) error {
   c.client.RefreshMetadata(topics...)//加载元数据
   sess, err := c.newSession(ctx, topics, handler, c.config.Consumer.Group.Rebalance.Retry.Max)
   go c.loopCheckPartitionNumbers(topics, sess)
}
复制

func (client *client) RefreshMetadata(topics ...string) error {
  return client.tryRefreshMetadata(topics, client.conf.Metadata.Retry.Max, deadline)
}
复制

func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int, deadline time.Time) error {
    broker = client.any()
    req := &MetadataRequest{Topics: topics, AllowAutoTopicCreation: allowAutoTopicCreation}
    response, err := broker.GetMetadata(req)
    shouldRetry, err := client.updateMetadata(response, allKnownMetaData)
}
复制

func (c *consumerGroup) newSession(ctx context.Context, topics []string, handler ConsumerGroupHandler, retries int) (*consumerGroupSession, error) {
    coordinator, err := c.client.Coordinator(c.groupID)
    join, err := c.joinGroupRequest(coordinator, topics)
    groupRequest, err := c.syncGroupRequest(coordinator, plan, join.GenerationId)
  return newConsumerGroupSession(ctx, c, claims, join.MemberId, join.GenerationId, handler)
}
复制

func newConsumerGroupSession(ctx context.Context, parent *consumerGroup, claims map[string][]int32, memberID string, generationID int32, handler ConsumerGroupHandler) (*consumerGroupSession, error) {
    offsets, err := newOffsetManagerFromClient(parent.groupID, memberID, generationID, parent.client)
    go sess.heartbeatLoop() 
      // start consuming
    for topic, partitions := range claims {
      for _, partition := range partitions {
        sess.waitGroup.Add(1)
        go func(topic string, partition int32) {
        sess.consume(topic, partition)
      }(topic, partition)
    }
  }
}
复制

type consumerGroupSession struct {
  parent       *consumerGroup
  memberID     string
  generationID int32
  handler      ConsumerGroupHandler


  claims  map[string][]int32
  offsets *offsetManager
  ctx     context.Context
  cancel  func()


  waitGroup       sync.WaitGroup
  releaseOnce     sync.Once
  hbDying, hbDead chan none
}
复制

func (s *consumerGroupSession) consume(topic string, partition int32) {
  // create new claim
  claim, err := newConsumerGroupClaim(s, topic, partition, offset)
   s.handler.ConsumeClaim(s, claim)
}
复制

func newConsumerGroupClaim(sess *consumerGroupSession, topic string, partition int32, offset int64) (*consumerGroupClaim, error) {
  pcm, err := sess.parent.consumer.ConsumePartition(topic, partition, offset)
}
复制

type consumerGroupClaim struct {
  topic     string
  partition int32
  offset    int64
  PartitionConsumer
}
复制

func (c *consumer) ConsumePartition(topic string, partition int32, offset int64) (PartitionConsumer, error) {
    child := &partitionConsumer
    if err := child.chooseStartingOffset(offset); err != nil
      if leader, err = c.client.Leader(child.topic, child.partition); err != nil
      if err := c.addChild(child); err != nil
        c.children[child.topic] = topicChildren
        topicChildren[child.partition] = child
      go withRecover(child.dispatcher)
      go withRecover(child.responseFeeder)
      child.broker = c.refBrokerConsumer(leader)
        bc := c.brokerConsumers[broker]
        bc.refs++
      child.broker.input <- child
}
复制

type partitionConsumer struct {
  highWaterMarkOffset int64  must be at the top of the struct because https://golang.org/pkg/sync/atomic/#pkg-note-BUG


  consumer *consumer
  conf     *Config
  broker   *brokerConsumer
  messages chan *ConsumerMessage
  errors   chan *ConsumerError
  feeder   chan *FetchResponse


  preferredReadReplica int32


  trigger, dying chan none
  closeOnce      sync.Once
  topic          string
  partition      int32
  responseResult error
  fetchSize      int32
  offset         int64
  retries        int32
}
复制

    func (child *partitionConsumer) dispatcher() 
      for range child.trigger
        if err := child.dispatch(); err != 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 
      broker, err := child.preferredBroker()
      child.broker = child.consumer.refBrokerConsumer(broker)
      child.broker.input <- child
复制

type brokerConsumer struct {
  consumer         *consumer
  broker           *Broker
  input            chan *partitionConsumer
  newSubscriptions chan []*partitionConsumer
  subscriptions    map[*partitionConsumer]none
  wait             chan none
  acks             sync.WaitGroup
  refs             int
}
复制

func (c *consumer) refBrokerConsumer(broker *Broker) *brokerConsumer {
   bc = c.newBrokerConsumer(broker)
}
复制

func (c *consumer) newBrokerConsumer(broker *Broker) *brokerConsumer 
      go withRecover(bc.subscriptionManager)
      go withRecover(bc.subscriptionConsumer)
复制

func (bc *brokerConsumer) subscriptionManager(){
      case event, ok := <-bc.input:
         buffer = append(buffer, event)
      case bc.newSubscriptions <- buffer
         buffer = nil
 }
复制

func (bc *brokerConsumer) subscriptionConsumer() 
      for newSubscriptions := range bc.newSubscriptions {
          bc.updateSubscriptions(newSubscriptions)
          response, err := bc.fetchNewMessages()
          bc.acks.Add(len(bc.subscriptions))
          child.feeder <- response
          bc.acks.Wait()
          bc.handleResponses()
复制

func (bc *brokerConsumer) fetchNewMessages() (*FetchResponse, error) {
     for child := range bc.subscriptions {
        request.AddBlock(child.topic, child.partition, child.offset, child.fetchSize)
     }
     return bc.broker.Fetch(request)
}
复制

func (b *Broker) Fetch(request *FetchRequest) (*FetchResponse, error) {
  response := new(FetchResponse)
  err := b.sendAndReceive(request, response)
}
复制

 func (child *partitionConsumer) responseFeeder() {
   feederLoop:
      从broker获取消息的大循环
      for response := range child.feeder
           for i, msg := range msgs {
               case child.messages <- msg:
                   child.broker.input <- child
                   continue feederLoop
 }
复制

func (child *partitionConsumer) Messages() <-chan *ConsumerMessage {
  return child.messages
}
复制