Kafka简单入门

1
2
3
4
5

# For Kraft mode
kafka-topics.sh --create --topic golang-test --partitions 3 --bootstrap-server localhost:9092

# For Zookeeper mode
# kafka-topics.sh --create --topic golang-test --partitions 3 --replication-factor 1 --zookeeper localhost:2181

1
2
3
4

kafka-console-producer.sh --topic golang-test --bootstrap-server localhost:9092
>Hello Kafka
>This is a message from CLI
>Learning Golang with Kafka

1

kafka-console-consumer.sh --topic golang-test --from-beginning --bootstrap-server localhost:9092

1

go get github.com/confluentinc/confluent-kafka-go/v2/kafka

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

// producer.go
package main

import (
	"fmt"
	"log"
	"strconv"
	"time"

	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
)

func main() {
	// --- Producer Configuration ---
	// Create a new producer instance with a configuration map.
	// 'bootstrap.servers' points to the Kafka broker(s).
	p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost:9092"})
	if err != nil {
		log.Fatalf("Failed to create producer: %s", err)
	}
	defer p.Close()

	// --- Delivery Report Goroutine ---
	// This goroutine handles delivery reports from Kafka.
	// It's crucial for knowing if messages were successfully sent.
	go func() {
		for e := range p.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					// Message failed to be delivered.
					fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
				} else {
					// Message was successfully delivered.
					fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
				}
			}
		}
	}()

	// --- Message Production ---
	// Define the topic to which we want to produce messages.
	topic := "golang-test"
	for i := 0; i < 10; i++ {
		// Prepare the message key and value.
		key := "key-" + strconv.Itoa(i)
		value := "hello world from golang producer - message " + strconv.Itoa(i)

		// Produce the message. This is an asynchronous operation.
		// The delivery report is handled by the goroutine above.
		err := p.Produce(&kafka.Message{
			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
			Key:            []byte(key),
			Value:          []byte(value),
		}, nil)

		if err != nil {
			log.Printf("Failed to produce message: %v\n", err)
		}
		
		// Wait for a short duration before sending the next message.
		time.Sleep(500 * time.Millisecond)
	}

	// --- Flush and Close ---
	// Wait for all messages in the producer's queue to be delivered.
	// The timeout is set to 15 seconds.
	p.Flush(15 * 1000)
	fmt.Println("Producer has finished sending messages.")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

// consumer.go
package main

import (
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
)

func main() {
	// --- Consumer Configuration ---
	// Create a new consumer instance with a configuration map.
	// 'bootstrap.servers' points to the Kafka broker(s).
	// 'group.id' is essential for consumer groups.
	// 'auto.offset.reset' determines the starting point when no offset is stored.
	// 'earliest' means start from the beginning of the topic.
	c, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": "localhost:9092",
		"group.id":          "my-golang-group",
		"auto.offset.reset": "earliest",
	})
	if err != nil {
		log.Fatalf("Failed to create consumer: %s", err)
	}
	defer c.Close()

	// --- Topic Subscription ---
	// Subscribe to the topic.
	topic := "golang-test"
	err = c.SubscribeTopics([]string{topic}, nil)
	if err != nil {
		log.Fatalf("Failed to subscribe to topic: %s", err)
	}

	// --- Graceful Shutdown Setup ---
	// Set up a channel to listen for OS signals (like Ctrl+C).
	sigchan := make(chan os.Signal, 1)
	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

	// --- Message Consumption Loop ---
	run := true
	for run {
		select {
		case sig := <-sigchan:
			// Received a shutdown signal.
			fmt.Printf("Caught signal %v: terminating\n", sig)
			run = false
		default:
			// Poll for a message. The timeout is set to 100 milliseconds.
			// This is a non-blocking call.
			ev := c.Poll(100)
			if ev == nil {
				continue // No event, continue polling.
			}

			switch e := ev.(type) {
			case *kafka.Message:
				// Received a message.
				fmt.Printf("Received message on %s: Key: %s | Value: %s\n", e.TopicPartition, string(e.Key), string(e.Value))
			case kafka.Error:
				// Encountered an error.
				// This could be a transient error or a fatal one.
				fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n", e.Code(), e)
				if e.IsFatal() {
					run = false
				}
			default:
				// Other event types (e.g., stats).
				fmt.Printf("Ignored event: %v\n", e)
			}
		}
	}
	fmt.Println("Consumer is shutting down.")
}