Kafka数据存储

# 安装kafka
$ helm repo add bitnami https://charts.bitnami.com/bitnami
$ helm install kafka-master bitnami/kafka -n ps

# 安装kafka-client
$ kubectl run kafka-master-client --restart='Never' --image docker.io/bitnami/kafka:3.2.0-debian-10-r4 --namespace ps --command -- sleep infinity

# 进入容器
$ kubectl exec --tty -i kafka-master-client --namespace ps -- bash

# 使用client调用kafka-server
# 设置环境变量kafka-server地址
$ export kafka=kafka-master-0.kafka-master-headless.ps.svc.cluster.local:9092

$ cd /opt/bitnami/kafka/bin
# 创建topic
$ kafka-topics.sh --create --topic test --bootstrap-server $kafka

# 启动 consumer-console
$ kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server $kafka

# 启动 producer-console
$ kafka-console-producer.sh --topic test --bootstrap-server $kafka

# 查看日志文件
$ /opt/bitnami/kafka/bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000000000000.log --print-data-log
$ /opt/bitnami/kafka/bin/kafka-dump-log.sh --files 00000000000000000000.log --print-data-log
# output:
baseOffset: 45 lastOffset: 45 count: 1 baseSequence: 0 lastSequence: 0 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 2031009 CreateTime: 1654672883918 size: 331 magic: 2 compresscodec: gzip crc: 4084402132 isvalid: true
| offset: 45 CreateTime: 1654672883918 keySize: -1 valueSize: 389 sequence: 0 headerKeys: [] payload: {"widget":{"debug":"on","window":{"title":"Sample Konfabulator Widget","name":"main_window","width":500,"height":500},"image":{"src":"Images/Sun.png","name":"sun1","hOffset":250,"vOffset":250,"alignment":"center"},"text":{"data":"Click Here","size":36,"style":"bold","name":"text1","hOffset":250,"vOffset":100,"alignment":"center","onMouseUp":"sun1.opacity = (sun1.opacity / 100) * 90;"}}}

# 查看索引文件
$ /opt/bitnami/kafka/bin/kafka-dump-log.sh --files 00000000000000000000.index
$ /opt/bitnami/kafka/bin/kafka-dump-log.sh --files 00000000000000000000.timeindex

// consumer.go
package kafka_demo

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

type KafkaClient struct {
	topics        []string
	consumerGroup sarama.ConsumerGroup
}

func NewKafkaClient(brokers string, topics string, groupID string, oldest bool, version string, assignor string, verbose bool) *KafkaClient {
	if verbose {
		sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
	}

	config := sarama.NewConfig()
	ver, err := sarama.ParseKafkaVersion(version)
	if err != nil {
		log.Panic(err)
	}
	config.Version = ver

	switch assignor {
	case "sticky":
		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky
	case "roundrobin":
		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
	case "range":
		config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
	default:
		log.Panicf("Unrecognized consumer group partition assignor: %s", assignor)
	}

	if oldest {
		config.Consumer.Offsets.Initial = sarama.OffsetOldest
	}
	kafkaClient := &KafkaClient{
		topics: strings.Split(topics, ","),
	}
	kafkaClient.consumerGroup, err = sarama.NewConsumerGroup(strings.Split(brokers, ","), groupID, config)
	if err != nil {
		log.Panic(err)
	}
	return kafkaClient
}

func (k *KafkaClient) Consume(handler Handler) {
	keepRunning := true
	log.Println("Starting a new Sarama consumer")

	/**
	 * Setup a new Sarama consumer group
	 */
	consumer := Consumer{
		ready:   make(chan bool),
		handler: handler,
	}

	ctx, cancel := context.WithCancel(context.Background())

	consumptionIsPaused := false
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			// `Consume` should be called inside an infinite loop, when a
			// server-side rebalance happens, the consumer session will need to be
			// recreated to get the new claims
			if err := k.consumerGroup.Consume(ctx, k.topics, &consumer); err != nil {
				log.Panicf("Error from consumer: %v", err)
			}
			// check if context was cancelled, signaling that the consumer should stop
			if ctx.Err() != nil {
				return
			}
			consumer.ready = make(chan bool)
		}
	}()

	<-consumer.ready // Await till the consumer has been set up
	log.Println("Sarama consumer up and running!...")

	sigusr1 := make(chan os.Signal, 1)
	signal.Notify(sigusr1, syscall.SIGUSR1)

	sigterm := make(chan os.Signal, 1)
	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)

	for keepRunning {
		select {
		case <-ctx.Done():
			log.Println("terminating: context cancelled")
			keepRunning = false
		case <-sigterm:
			log.Println("terminating: via signal")
			keepRunning = false
		case <-sigusr1:
			toggleConsumptionFlow(k.consumerGroup, &consumptionIsPaused)
		}
	}
	cancel()
	wg.Wait()
	if err := k.consumerGroup.Close(); err != nil {
		log.Panicf("Error closing client: %v", err)
	}
}

func toggleConsumptionFlow(client sarama.ConsumerGroup, isPaused *bool) {
	if *isPaused {
		client.ResumeAll()
		log.Println("Resuming consumption")
	} else {
		client.PauseAll()
		log.Println("Pausing consumption")
	}

	*isPaused = !*isPaused
}

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

// Setup is run at the beginning of a new session, before ConsumeClaim
func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
	// Mark the consumer as ready
	close(consumer.ready)
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	// NOTE:
	// Do not move the code below to a goroutine.
	// The `ConsumeClaim` itself is called within a goroutine, see:
	// https://github.com/Shopify/sarama/blob/main/consumer_group.go#L27-L29
	for {
		select {
		case message := <-claim.Messages():
			log.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
			err := consumer.handler.Handler(message)
			if err != nil {
				log.Println(err)
			} else {
				session.MarkMessage(message, "")
			}

		// Should return when `session.Context()` is done.
		// If not, will raise `ErrRebalanceInProgress` or `read tcp <ip>:<port>: i/o timeout` when kafka rebalance. see:
		// https://github.com/Shopify/sarama/issues/1192
		case <-session.Context().Done():
			return nil
		}
	}
}

type Handler interface {
	Handler(message *sarama.ConsumerMessage) error
}

// producer.go
package kafka_demo

import (
	"github.com/Shopify/sarama"
	"strings"
	"time"
)

type kafkaSyncProducer struct {
	Topic        string
	syncProducer sarama.SyncProducer
}

func NewKafkaSyncProducer(brokers string, topic string) (*kafkaSyncProducer, error) {
	pr, err := sarama.NewSyncProducer(strings.Split(brokers, ","), nil)
	return &kafkaSyncProducer{
		Topic:        topic,
		syncProducer: pr,
	}, err
}

func (k *kafkaSyncProducer) SendMessages(messages [][]byte) (err error) {
	producerMessages := make([]*sarama.ProducerMessage, 0, 1)
	for _, msg := range messages {
		producerMessages = append(producerMessages, &sarama.ProducerMessage{
			Topic:     k.Topic,
			Key:       nil,
			Value:     sarama.ByteEncoder(msg),
			Headers:   nil,
			Metadata:  nil,
			Offset:    0,
			Partition: 0,
			Timestamp: time.Time{},
		})
	}
	if len(producerMessages) > 0 {
		err = k.syncProducer.SendMessages(producerMessages)
	}
	return
}

func (k *kafkaSyncProducer) SendMessage(message []byte) (partition int32, offset int64, err error) {
	partition, offset, err = k.syncProducer.SendMessage(&sarama.ProducerMessage{
		Topic:     k.Topic,
		Key:       nil,
		Value:     sarama.ByteEncoder(message),
		Headers:   nil,
		Metadata:  nil,
		Offset:    0,
		Partition: 0,
		Timestamp: time.Time{},
	})
	return
}

// size.go
package kafka_demo

import (
	"fmt"
	"math/rand"
	"time"
	"unsafe"
)

//
func newMessage() {
	//fmt.Println(unsafe.Sizeof([128]int64{}))
	msg1Kb := new1KbMessage()
	fmt.Printf("1KB: size = %d byte\n", unsafe.Sizeof(msg1Kb))

	msg1Mb := new100KbMessage()
	fmt.Printf("100KB: size = %d byte\n", unsafe.Sizeof(msg1Mb))
}

func new1KbMessage() [1024]byte {
	v := [1024]byte{}
	return v
}

func new100KbMessage() [102400]byte {
	v := [102400]byte{}
	for i := 0; i < 102400; i++ {
		rand.Seed(time.Now().UnixNano())
		v[i] = byte(rand.Intn(100))
	}
	return v
}

# producer_test.go
package kafka_demo

import (
	"strconv"
	"testing"
)

func newDefaultProducer() *kafkaSyncProducer {
	brokers := "kafka-master-0.kafka-master-headless.ps.svc.cluster.local:9092"
	topics := "test"
	ps, _ := NewKafkaSyncProducer(brokers, topics)
	return ps
}

func TestSyncProducer(t *testing.T) {
	brokers := "kafka-master-0.kafka-master-headless.ps.svc.cluster.local:9092"
	topics := "test"
	ps, _ := NewKafkaSyncProducer(brokers, topics)
	for i := 0; i < 16; i++ {
		val := []byte(strconv.FormatInt(int64(i), 10))
		_, _, err := ps.SendMessage(val)
		if err != nil {
			t.Error(err)
		} else {
			//t.Logf("produce Message: value = %s, partition = %v, offset = %v", string(val), partition, offset)
		}
	}
}

func TestSyncProducer1Kb(t *testing.T) {
	producer := newDefaultProducer()
	m := new1KbMessage()
	counter := 1024 * 50
	for i := 0; i < counter; i++ {
		_, _, err := producer.SendMessage(m[:])
		if err != nil {
			t.Error(err)
		}
	}
	producer.syncProducer.Close()
}

func TestSyncProducer500KB(t *testing.T) {
	producer := newDefaultProducer()
	m := new100KbMessage()
	counter := 512
	for i := 0; i < counter; i++ {
		_, _, err := producer.SendMessage(m[:])
		if err != nil {
			t.Error(err)
		}
	}
	producer.syncProducer.Close()
}

func TestSyncProducerString(t *testing.T) {
	producer := newDefaultProducer()
	m := []byte(`{"name": "Bob", "age": 20}`)
	counter := 512
	for i := 0; i < counter; i++ {
		_, _, err := producer.SendMessage(m[:])
		if err != nil {
			t.Error(err)
		}
	}
	producer.syncProducer.Close()
}

# 50MB消息，单条消息100KB
# 写入前
0       00000000000000000000.index
0       00000000000000000000.log
0       00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

#写入后
8.0K    00000000000000000000.index
43M     00000000000000000000.log
8.0K    00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

# 50MB消息，单条消息100KB
# 写入前
8.0K    00000000000000000000.index
43M     00000000000000000000.log
8.0K    00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

# 写入后
12K     00000000000000000000.index
85M     00000000000000000000.log
16K     00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

# 50MB消息，单条消息100KB
# 写入前
28K     00000000000000000000.index
277M    00000000000000000000.log
40K     00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

# 写入后
32K     00000000000000000000.index
327M    00000000000000000000.log
44K     00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

# 50MB消息，单条消息100KB
# 写入前
32K     00000000000000000000.index
327M    00000000000000000000.log
44K     00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

# 写入后
36K     00000000000000000000.index
377M    00000000000000000000.log
52K     00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

# 50MB消息，单条消息100KB
# 写入前
24K     00000000000000000000.index
235M    00000000000000000000.log
32K     00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata

# 写入后
28K     00000000000000000000.index
277M    00000000000000000000.log
40K     00000000000000000000.timeindex
4.0K    leader-epoch-checkpoint
4.0K    partition.metadata