Kafka體系架構、命令、Go案例

2020 年 12 月 30 日
筆記
golang

原文地址：//github.com/WilburXu/blog/blob/master/kafka/Kafka基本架構和命令.md

Kafka體系架構

Broker服務代理節點

服務代理節點。對於Kafka而言，Broker可以簡單地看作一個獨立的Kafka服務節點或Kafka服務實例。大多數情況下也可以將Broker看作一台Kafka服務器，前提是這台服務器上只部署了一個Kafka實例，一個或多個Broker組成了一個Kafka集群。

Producer和Consumer

Producer生產者

生產者，也就是發送消息的一方。生產者負責創建消息，然後將其投遞到Kafka中。

一個正常的生產邏輯需要具備以下幾個步驟：

創建生產者實例
構建待發送的消息
發送消息到指定的Topic、Partition、Key
關閉生產者實例

Consumer消費者

消費者，也就是接收消息的一方。消費者連接到Kafka上並接收消息，從而進行相應的業務邏輯處理。

消費一般有三種消費模式：

單線程模式

單個線程消費多個Partition

問題：

效率低，並發上不去
可用性差，單個線程掛了，將無法消費

多線程模式

獨立消費者模式

和單線程模式類似，區別就是為每一個Partition單獨起一個線程進行消費。

問題：

線程和並發增加了，但是單線程掛了，該線程的分區還是無法消費。

消費組模式

也是目前最常用的消費模式，我們可以創建多個消費實例並設置同一個group-id來區分消費組，同一個消費組可以指定一個或多個Topic進行消費：

消費組自平衡（Rebalance），kafka會根據消費組實例數量和分區數量自平衡分配
不會重複消費，同個組內kafka確保一個分區只會發往一個消費實例，避免重複消費
高可用，當一個消費實例掛了，kafka會自動調整消費實例和分區的關係

Topic主題

Kafka中的消息以主題為單位進行歸類（邏輯概念，生產者負責將消息發送到特定的主題（發送到Kafka集群中的每一條消息都要指定一個主題），而消費者負責訂閱主題並進行消費。

Partition分區

物理分區，主題細分為了1或多個分區，一個分區只能屬於單個主題，一般也會把分區稱為主題分區（Topic-Partition）。

Segment

實際存儲數據的地方，Segment包含一個數據文件和一個索引文件。一個Partition有多個大小相同的Segment，可以理解為Partition是在Segment之上進行的邏輯抽象。

Kafka基本命令

zookeeper

broker節點保存在zookeeper，所有需要：

進入zookeeper，然後 ./bin/zkCli.sh
執行ls /brokers/ids

查看broker詳情

kafka-log-dirs.sh --describe --bootstrap-server kafka:9092 --broker-list 1

topic

查看列表

kafka-topics.sh --list --zookeeper zookeeper:2181

創建

kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 3 --topic [topic_name]

查看詳情

kafka-topics.sh --describe --zookeeper zookeeper:2181 --topic [topic_name]

刪除

kafka-topics.sh --zookeeper zookeeper:2181 --delete --topic [topic_name]

topic消費情況

topic offset 最小

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 -topic [topic_name] --time -2

topic offset最大

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 -topic [topic_name] --time -1

生產

添加數據

kafka-console-producer.sh --broker-list localhost:9092 --topic [topic_name]

消費

從頭部開始消費

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic [topic_name] --from-beginning

從尾部開始消費，必需要指定分區

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic [topic_name] --offset latest --partition 0

從某個位置開始消費(–offset [n])

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic [topic_name] --offset 100 --partition 0

消費指定個數(–max-messages [n])

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic [topic_name] --offset latest --partition 0 --max-messages 2

消費組

查看消費組列表

kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

查看消費組情況

kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group [group_id]

offset 偏移設置為最早

kafka-consumer-groups.bat --bootstrap-server kafka:9092 --group kafka_consumer_session --reset-offsets --to-earliest --all-topics --execute

offset 偏移設置為新

kafka-consumer-groups.bat --bootstrap-server kafka:9092 --group kafka_consumer_session --reset-offsets --to-latest --all-topics --execute

offset 偏移設置為指定位置

kafka-consumer-groups.bat --bootstrap-server kafka:9092 --group kafka_consumer_session --reset-offsets --to-offset 2000 --all-topics --execute

offset 偏移設置某個時間之後最早位移

kafka-consumer-groups.bat --bootstrap-server kafka:9092 --group kafka_consumer_session --reset-offsets --to-datetime 2020-12-28T00:00:00.000 --all-topics --execute

Go案例

基於//github.com/Shopify/sarama的生產和消費案例

生產者

InitKafka.go

package kafka

var (
	kafkaClient *Client
)

func InitKafka() {
	var err error

	var config = Config{
		Host: []string{"kafka:9092"},
	}

	kafkaClient, err = NewClient(config)
	if err != nil {
		panic(err)
	}
}

func GetClient() *Client {
	return kafkaClient
}

Producer.go

package kafka

import (
   "errors"
   "github.com/Shopify/sarama"
)

type Client struct {
   sarama.AsyncProducer
   msgPool chan *sarama.ProducerMessage
}

type Config struct {
   Host          []string `json:"host"`
   ReturnSuccess bool     `json:"return_success"`
   ReturnErrors  bool     `json:"return_errors"`
}

func NewClient(cfg Config) (*Client, error) {
   // create client
   var err error
   c := &Client{
      msgPool: make(chan *sarama.ProducerMessage, 2000),
   }

   config := sarama.NewConfig()
   config.Producer.Return.Errors = cfg.ReturnErrors
   config.Producer.Return.Successes = cfg.ReturnSuccess
   config.Version = sarama.V2_0_0_0

   c.AsyncProducer, err = sarama.NewAsyncProducer(cfg.Host, config)
   if err != nil {
      return nil, err
   }

   return c, nil
}

// run
func (c *Client) Run() {
   for {
      select {
      case msg := <-c.msgPool:
         c.Input() <- msg
         logger.Info("%+v", msg)
      }
   }
}

// send msg
func (c *Client) Send(topic string, msg []byte) error {
   if topic == "" {
      return errors.New("kafka producer send msg topic empty")
   }

   kafkaMsg := &sarama.ProducerMessage{
      Topic: topic,
      Value: sarama.ByteEncoder(msg),
   }

   c.msgPool <- kafkaMsg

   return nil
}

生產者初始化：

// kafka init
kafka.InitKafka()
go kafka.GetClient().Run()

消費者

consumer.go

package kafka_consumer

import (
   "context"
   "github.com/Shopify/sarama"
   "os"
   "os/signal"
   "sync"
   "syscall"
)

// Consumer represents a Sarama consumer group consumer
type Consumer struct {
   ready chan bool
}

func (c *Consumer) Setup(session sarama.ConsumerGroupSession) error {
   //panic("implement me")
   return nil
}

func (c *Consumer) Cleanup(session sarama.ConsumerGroupSession) error {
   //panic("implement me")
   return nil
}

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
   for message := range claim.Messages() {
      logger.Info("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
      session.MarkMessage(message, "")
      c.Handler(message.Topic, message.Value)
   }
   
   return nil
}

func (c *Consumer) Handler(topic string, msg []byte) {
   switch topic {
   case conscom.KafkaTopicGiftOrder:
      GiftOrder(topic, msg)
   case conscom.KafkaTopicFollow:
      UserFollow(topic, msg)
   }
}

func ConsumeInit(topics []string, groupID string) {
   consumer := Consumer{
      ready: make(chan bool),
   }

   brokerList := []string{"kafka:9092"}

   config := sarama.NewConfig()
   config.Version = sarama.V1_0_0_0

   ctx, cancel := context.WithCancel(context.Background())
   client, err := sarama.NewConsumerGroup(brokerList, groupID, config)
   if err != nil {
      log.Printf("kafka consumer err %v", err)
      return
   }

   wg := &sync.WaitGroup{}
   wg.Add(1)
   go func() {
      defer wg.Done()
      for {
         // server-side rebalance happens, the consumer session will need to be
         if err := client.Consume(ctx, topics, &consumer); err != nil {
            log.Printf("kafka consumer: %v", err)
         }

         // check if context was cancelled, signaling that the consumer should stop
         if ctx.Err() != nil {
            return
         }
         consumer.ready = make(chan bool)
      }
   }()

   sigterm := make(chan os.Signal, 1)
   signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
   select {
   case <-ctx.Done():
      log.Printf("kafka consume gift terminating: context cancelled")
   case <-sigterm:
      log.Printf("kafka consume gift terminating: via signal")
   }
   cancel()
   wg.Wait()
   if err = client.Close(); err != nil {
      log.Printf("kafka consume gift Error closing client: %v", err)
   }
}

消費者初始化：

// kafka consumer
go kafka_consumer.ConsumeInit([]string{"topicA", "topicB", "group-name")

參考

《深入理解Kafka:核心設計與實踐原理》作者:朱忠華

//github.com/Shopify/sarama

//kafka.apache.org/documentation/

//crossoverjie.top/2018/11/20/kafka/kafka-consumer/

Tags: golang