docker run -d \
  --name kafka \
  -p 14818:9092 \
  -p 9093:9093 \
  -v /tmp/kraft-combined-logs:/tmp/kraft-combined-logs \
  -e TZ=Asia/Shanghai \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.16.11.111:14818 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
  -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
  -e KAFKA_NUM_PARTITIONS=3 \
  -e KAFKA_LOG_DIRS=/tmp/kraft-combined-logs \
  -e CLUSTER_ID=5L6g3nShT-eMCtK--X86sw \
  apache/kafka-native:4.1.0

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: kafka
  name: kafka
  namespace: moonfdd
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      labels:
        app: kafka
    spec:
      initContainers:
        - name: kafka-fix-data-volume-permissions
          image: alpine
          imagePullPolicy: IfNotPresent
          command:
          - sh
          - -c
          - "chown -R 1000:1000 /tmp/kraft-combined-logs"
          volumeMounts:
            - mountPath: /tmp/kraft-combined-logs
              name: volv
      containers:
        - env:
            - name: TZ
              value: Asia/Shanghai
            - name: KAFKA_NODE_ID
              value: "1"
            - name: KAFKA_PROCESS_ROLES
              value: broker,controller
            - name: KAFKA_LISTENERS
              value: PLAINTEXT://:9092,CONTROLLER://:9093
            - name: KAFKA_ADVERTISED_LISTENERS
              value: PLAINTEXT://172.16.11.111:14818
            - name: KAFKA_CONTROLLER_LISTENER_NAMES
              value: CONTROLLER
            - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
              value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
            - name: KAFKA_CONTROLLER_QUORUM_VOTERS
              value: 1@localhost:9093
            - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
              value: "1"
            - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR
              value: "1"
            - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR
              value: "1"
            - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS
              value: "0"
            - name: KAFKA_NUM_PARTITIONS
              value: "3"
            - name: KAFKA_LOG_DIRS
              value: /tmp/kraft-combined-logs
            - name: CLUSTER_ID
              value: "5L6g3nShT-eMCtK--X86sw"  # 固定集群ID，僅首次啟動格式化使用
          image: 'apache/kafka-native:4.1.0'
          imagePullPolicy: IfNotPresent
          name: kafka
          volumeMounts:
            - mountPath: /tmp/kraft-combined-logs
              name: volv
      volumes:
        - hostPath:
            path: /root/k8s/moonfdd/kafka/tmp/kraft-combined-logs
            type: DirectoryOrCreate
          name: volv
---
apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka
  name: kafka
  namespace: moonfdd
spec:
  ports:
    - port: 9092
      protocol: TCP
      targetPort: 9092
      name: 9092-9092
    - port: 9093
      protocol: TCP
      targetPort: 9093
      name: 9093-9093
  selector:
    app: kafka
  type: NodePort

package main

import (
	"context"
	"log"

	"github.com/segmentio/kafka-go"
)

func main() {
	// 創(chuàng)建一個Kafka writer（Producer）
	w := kafka.NewWriter(kafka.WriterConfig{
		Brokers:  []string{"172.16.11.111:14818"}, // Kafka broker 地址
		Topic:    "test-topic",                    // 發(fā)送的 topic
		Balancer: &kafka.LeastBytes{},             // 負(fù)載均衡策略
	})

	// 寫入消息
	err := w.WriteMessages(context.Background(),
		kafka.Message{
			Key:   []byte("Key-A"),
			Value: []byte("Hello Kafka from Go!"),
		},
	)

	if err != nil {
		log.Fatalf("could not write message: %v", err)
	}

	log.Println("Message sent successfully!")

	// 關(guān)閉 writer
	w.Close()
}

package main

import (
	"context"
	"log"

	"github.com/segmentio/kafka-go"
)

func main() {
	// 創(chuàng)建 Kafka reader（Consumer）
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  []string{"172.16.11.111:14818"}, // Kafka broker 地址
		Topic:    "test-topic",                    // 訂閱的 topic
		GroupID:  "my-consumer-group",             // 消費(fèi)者組，確保相同組會讀取上一 offset
		MinBytes: 10e3,                            // 最小fetch字節(jié)數(shù)
		MaxBytes: 10e6,                            // 最大fetch字節(jié)數(shù)
	})

	for {
		// 讀取消息（會自動從上次的 offset 開始）
		m, err := r.ReadMessage(context.Background())
		if err != nil {
			log.Fatalf("could not read message: %v", err)
		}
		log.Printf("offset:%d | key:%s | value:%s\n", m.Offset, string(m.Key), string(m.Value))
	}

	// r.Close() // 如果你打算退出循環(huán)時關(guān)閉
}

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/IBM/sarama"
)

func main() {
	// 配置生產(chǎn)者
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true          // 確保消息發(fā)送成功
	config.Producer.RequiredAcks = sarama.WaitForAll // 等待所有副本確認(rèn)
	config.Producer.Retry.Max = 3                    // 重試次數(shù)

	// 重要：配置客戶端使用正確的主機(jī)
	config.Net.SASL.Enable = false
	config.Net.TLS.Enable = false
	config.Version = sarama.MaxVersion

	// 創(chuàng)建同步生產(chǎn)者
	producer, err := sarama.NewSyncProducer([]string{"172.16.11.111:14818"}, config)
	if err != nil {
		log.Fatalf("創(chuàng)建生產(chǎn)者失敗: %v", err)
	}
	defer producer.Close()

	// 構(gòu)造消息
	message := &sarama.ProducerMessage{
		Topic: "test-topic",
		Key:   sarama.StringEncoder("message-key"),
		Value: sarama.StringEncoder(fmt.Sprintf("Hello Kafka! %v", time.Now())),
	}

	// 發(fā)送消息
	partition, offset, err := producer.SendMessage(message)
	if err != nil {
		log.Fatalf("發(fā)送消息失敗: %v", err)
	}

	fmt.Printf("消息發(fā)送成功! 分區(qū): %d, 偏移量: %d\n", partition, offset)
}

package main

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

	"github.com/IBM/sarama"
)

type Consumer struct{}

func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
	// 會話初始化，可以在這里做一些準(zhǔn)備工作
	return nil
}

func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	// 會話結(jié)束時的清理操作
	return nil
}

func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	// claim.Messages() 會不斷返回新消息
	for msg := range claim.Messages() {
		fmt.Printf("Topic:%s Partition:%d Offset:%d Value:%s\n",
			msg.Topic, msg.Partition, msg.Offset, string(msg.Value))

		// 標(biāo)記該消息已被處理，Kafka會自動保存offset
		session.MarkMessage(msg, "")
	}
	return nil
}

func main() {
	// Kafka集群地址
	brokers := []string{"172.16.11.111:14818"}
	groupID := "my-group" // 消費(fèi)者組ID，保持不變才能從上次offset消費(fèi)
	topics := []string{"test-topic"}

	// 配置
	config := sarama.NewConfig()
	config.Version = sarama.MaxVersion // Kafka版本
	config.Consumer.Return.Errors = true

	// 非首次啟動時自動從上次位置開始
	config.Consumer.Offsets.Initial = sarama.OffsetNewest
	// OffsetNewest: 如果沒有歷史offset，從最新開始；
	// OffsetOldest: 如果沒有歷史offset，從最舊開始。

	// 創(chuàng)建消費(fèi)者組
	consumerGroup, err := sarama.NewConsumerGroup(brokers, groupID, config)
	if err != nil {
		log.Fatalf("Error creating consumer group: %v", err)
	}
	defer consumerGroup.Close()

	consumer := &Consumer{}

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

	go func() {
		for err := range consumerGroup.Errors() {
			log.Printf("Error: %v", err)
		}
	}()

	log.Println("Kafka consumer started...")
	// 優(yōu)雅退出
	go func() {
		sigchan := make(chan os.Signal, 1)
		signal.Notify(sigchan, os.Interrupt)
		<-sigchan
		cancel()
	}()

	// 循環(huán)消費(fèi)
	for {
		if err := consumerGroup.Consume(ctx, topics, consumer); err != nil {
			log.Printf("Error from consumer: %v", err)
		}

		// 檢查退出
		if ctx.Err() != nil {
			return
		}
	}
}