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

1package main
2
3import (
4"fmt"
5"github.com/confluentinc/confluent-kafka-go/kafka"
6)
7
8func main() {
9
10	var kafkaconf = &kafka.ConfigMap{
11		// 接入点
12		"bootstrap.servers": "接入点",
13		// 接入协议
14		"security.protocol": "sasl_ssl",
15		// 证书文件路径
16		"ssl.ca.location": "ca.pem",
17		// SASL 机制
18		"sasl.mechanism": "SCRAM-SHA-512",
19		// SASL 用户名
20		"sasl.username": "alice",
21		// SASL 用户密码
22		"sasl.password": "alice1234!",
23	}
24
25	p, err := kafka.NewProducer(kafkaconf)
26	if err != nil {
27		panic(err)
28	}
29
30	defer p.Close()
31
32	// Delivery report handler for produced messages
33	go func() {
34		for e := range p.Events() {
35			switch ev := e.(type) {
36			case *kafka.Message:
37				if ev.TopicPartition.Error != nil {
38					fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
39				} else {
40					fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
41				}
42			}
43		}
44	}()
45
46	// 填写创建的主题名称
47	topic := "topic_name"
48	for _, word := range []string{"Golang", "for", "kafka", "client", "test"} {
49		p.Produce(&kafka.Message{
50			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
51			Value:          []byte(word),
52		}, nil)
53	}
54
55	// Wait for message deliveries before shutting down
56	p.Flush(15 * 1000)
57}

1package main
2
3import (
4"fmt"
5"github.com/confluentinc/confluent-kafka-go/kafka"
6)
7
8
9func main() {
10
11	var kafkaconf = &kafka.ConfigMap{
12		// 接入点
13		"bootstrap.servers": "接入点",
14		// 接入协议
15		"security.protocol": "sasl_ssl",
16		// 证书文件路径
17		"ssl.ca.location": "ca.pem",
18		// SASL 机制
19		"sasl.mechanism": "SCRAM-SHA-512",
20		// SASL 用户名
21		"sasl.username": "alice",
22		// SASL 用户密码
23		"sasl.password": "alice1234!",
24		// 消费组 id
25		"group.id": "test_group",
26		"auto.offset.reset": "earliest",
27	}
28
29	c, err := kafka.NewConsumer(kafkaconf)
30
31	if err != nil {
32		panic(err)
33	}
34
35	c.SubscribeTopics([]string{"topic_name"}, nil)
36
37	for {
38		msg, err := c.ReadMessage(-1)
39		if err == nil {
40			fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
41		} else {
42			// The client will automatically try to recover from all errors.
43			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
44		}
45	}
46
47	c.Close()
48}

1# 启动消费者
2go consumer.go
3# 启动生产者
4go producer.go