剖析Kafka Go语言客户端Sarama——ConsumerGroupSession到ConsumerGroupClaim的信令

后端

2023-09-03 00:43:33

写在前面

上一章中，我们初步介绍了Sarama consumergroup的结构。从本章开始，我们将循序渐进，深入其底层的各个组件。本章先介绍从ConsumerGroupSession到ConsumerGroupClaim的信令。

信令概述

在Sarama中，ConsumerGroupSession和ConsumerGroupClaim之间存在着信令机制，用于协调消费者之间的通信和协作。信令是通过发送和接收消息来实现的，这些消息称为信令消息。

信令消息分为两种类型：

JoinGroupRequest：当消费者加入消费者组时，它会向协调器发送JoinGroupRequest消息。该消息包含消费者组的名称、消费者的ID、消费者的会话超时时间以及消费者的订阅主题列表。
JoinGroupResponse：协调器收到JoinGroupRequest消息后，会向消费者发送JoinGroupResponse消息。该消息包含消费者组的成员列表、消费者的分配主题列表以及消费者组的协议版本。

信令流程

消费者加入消费者组的信令流程如下：

消费者向协调器发送JoinGroupRequest消息。
协调器收到JoinGroupRequest消息后，会将消费者添加到消费者组中，并向消费者发送JoinGroupResponse消息。
消费者收到JoinGroupResponse消息后，会根据消息中的分配主题列表，开始消费这些主题中的消息。

消费者离开消费者组的信令流程如下：

消费者向协调器发送LeaveGroupRequest消息。
协调器收到LeaveGroupRequest消息后，会将消费者从消费者组中移除，并向消费者发送LeaveGroupResponse消息。
消费者收到LeaveGroupResponse消息后，会停止消费消息。

信令示例

以下是一个消费者加入消费者组的信令示例：

import (
	"context"
	"fmt"
	"time"

	"github.com/Shopify/sarama"
)

func main() {
	// Create a new Sarama client.
	client, err := sarama.NewClient([]string{"localhost:9092"}, nil)
	if err != nil {
		panic(err)
	}

	// Create a new consumer group.
	consumerGroup, err := sarama.NewConsumerGroupFromClient("my-group", client)
	if err != nil {
		panic(err)
	}

	// Join the consumer group.
	err = consumerGroup.Join("my-topic", "my-consumer")
	if err != nil {
		panic(err)
	}

	// Consume messages from the consumer group.
	for {
		select {
		case msg := <-consumerGroup.Messages():
			fmt.Println("Received message:", string(msg.Value))
			consumerGroup.CommitUpto(msg)
		case err := <-consumerGroup.Errors():
			fmt.Println("Received error:", err)
		case <-time.After(time.Second * 10):
			fmt.Println("Timeout")
			consumerGroup.Leave()
			break
		}
	}

	// Leave the consumer group.
	consumerGroup.Leave()
}