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

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

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