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": "ssl",
15		// 证书文件路径
16		"ssl.ca.location": "ca.pem",
17        "ssl.certificate.location": "client.pem",
18        "ssl.key.location": "client.key",
19	}
20
21	p, err := kafka.NewProducer(kafkaconf)
22	if err != nil {
23		panic(err)
24	}
25
26	defer p.Close()
27
28	// Delivery report handler for produced messages
29	go func() {
30		for e := range p.Events() {
31			switch ev := e.(type) {
32			case *kafka.Message:
33				if ev.TopicPartition.Error != nil {
34					fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
35				} else {
36					fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
37				}
38			}
39		}
40	}()
41
42	// 填写创建的主题名称
43	topic := "topic_name"
44	for _, word := range []string{"Golang", "for", "kafka", "client", "test"} {
45		p.Produce(&kafka.Message{
46			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
47			Value:          []byte(word),
48		}, nil)
49	}
50
51	// Wait for message deliveries before shutting down
52	p.Flush(15 * 1000)
53}

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": "ssl",
16		// 证书文件路径
17		"ssl.ca.location": "ca.pem",
18        "ssl.certificate.location": "client.pem",
19        "ssl.key.location": "client.key",
20		// 消费组 id
21		"group.id": "test_group",
22		"auto.offset.reset": "earliest",
23	}
24
25	c, err := kafka.NewConsumer(kafkaconf)
26
27	if err != nil {
28		panic(err)
29	}
30
31	c.SubscribeTopics([]string{"topic_name"}, nil)
32
33	for {
34		msg, err := c.ReadMessage(-1)
35		if err == nil {
36			fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
37		} else {
38			// The client will automatically try to recover from all errors.
39			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
40		}
41	}
42
43	c.Close()
44}

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