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

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_plaintext",
16		// SASL 机制
17		"sasl.mechanism": "SCRAM-SHA-512",
18		// SASL 用户名
19		"sasl.username": "alice",
20		// SASL 用户密码
21		"sasl.password": "alice1234!",
22		// 消费组 id
23		"group.id": "test_group",
24		"auto.offset.reset": "earliest",
25	}
26
27	c, err := kafka.NewConsumer(kafkaconf)
28
29	if err != nil {
30		panic(err)
31	}
32    //填写创建的主题的名称
33	c.SubscribeTopics([]string{"topic_name"}, nil)
34
35	for {
36		msg, err := c.ReadMessage(-1)
37		if err == nil {
38			fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
39		} else {
40			// The client will automatically try to recover from all errors.
41			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
42		}
43	}
44
45	c.Close()
46}

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