This section takes Linux CentOS as an example to describe how to access a Kafka instance using a Kafka client in Go 1.16.5, including how to obtain the demo code, and produce and consume messages.
Before getting started, ensure that you have collected the information listed in Collecting Connection Information.
go version
If the following information is displayed, Go has been installed.
[root@ecs-test confluent-kafka-go]# go version go version go1.16.5 linux/amd64
If Go is not installed, do as follows to install it:
# Download the Go installation package. wget https://go.dev/dl/go1.16.5.linux-amd64.tar.gz # Decompress it to the /usr/local directory. The /usr/local directory can be changed as required. sudo tar -C /usr/local -xzf go1.16.5.linux-amd64.tar.gz # Set the environment variable. echo 'export PATH=$PATH:/usr/local/go/bin' >> ~/.profile source ~/.profile
go get github.com/confluentinc/confluent-kafka-go/kafka
package main
import (
"bufio"
"fmt"
"github.com/confluentinc/confluent-kafka-go/kafka"
"log"
"os"
"os/signal"
"syscall"
)
var (
brokers = "ip1:port1,ip2:port2,ip3:port3"
topics = "topic_name"
user = "username"
password = "password"
caFile = "phy_ca.crt" // Obtain the SSL certificate by referring to section "Collecting Connection Information". If Security Protocol is set to SASL_PLAINTEXT, delete this parameter.
)
func main() {
log.Println("Starting a new kafka producer")
config := &kafka.ConfigMap{
"bootstrap.servers": brokers,
"security.protocol": "SASL_SSL",
"sasl.mechanism": "PLAIN",
"sasl.username": user,
"sasl.password": password,
"ssl.ca.location": caFile, // If Security Protocol is set to SASL_PLAINTEXT, delete this parameter.
"ssl.endpoint.identification.algorithm": "none"
}
producer, err := kafka.NewProducer(config)
if err != nil {
log.Panicf("producer error, err: %v", err)
return
}
go func() {
for e := range producer.Events() {
switch ev := e.(type) {
case *kafka.Message:
if ev.TopicPartition.Error != nil {
log.Printf("Delivery failed: %v\n", ev.TopicPartition)
} else {
log.Printf("Delivered message to %v\n", ev.TopicPartition)
}
}
}
}()
// Produce messages to topic (asynchronously)
fmt.Println("please enter message:")
go func() {
for {
err := producer.Produce(&kafka.Message{
TopicPartition: kafka.TopicPartition{Topic: &topics, Partition: kafka.PartitionAny},
Value: GetInput(),
}, nil)
if err != nil {
log.Panicf("send message fail, err: %v", err)
return
}
}
}()
sigterm := make(chan os.Signal, 1)
signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
select {
case <-sigterm:
log.Println("terminating: via signal")
}
// Wait for message deliveries before shutting down
producer.Flush(15 * 1000)
producer.Close()
}
func GetInput() []byte {
reader := bufio.NewReader(os.Stdin)
data, _, _ := reader.ReadLine()
return data
}
The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.
package main
import (
"bufio"
"fmt"
"github.com/confluentinc/confluent-kafka-go/kafka"
"log"
"os"
"os/signal"
"syscall"
)
var (
brokers = "ip1:port1,ip2:port2,ip3:port3"
topics = "topic_name"
)
func main() {
log.Println("Starting a new kafka producer")
config := &kafka.ConfigMap{
"bootstrap.servers": brokers,
}
producer, err := kafka.NewProducer(config)
if err != nil {
log.Panicf("producer error, err: %v", err)
return
}
go func() {
for e := range producer.Events() {
switch ev := e.(type) {
case *kafka.Message:
if ev.TopicPartition.Error != nil {
log.Printf("Delivery failed: %v\n", ev.TopicPartition)
} else {
log.Printf("Delivered message to %v\n", ev.TopicPartition)
}
}
}
}()
// Produce messages to topic (asynchronously)
fmt.Println("please enter message:")
go func() {
for {
err := producer.Produce(&kafka.Message{
TopicPartition: kafka.TopicPartition{Topic: &topics, Partition: kafka.PartitionAny},
Value: GetInput(),
}, nil)
if err != nil {
log.Panicf("send message fail, err: %v", err)
return
}
}
}()
sigterm := make(chan os.Signal, 1)
signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
select {
case <-sigterm:
log.Println("terminating: via signal")
}
// Wait for message deliveries before shutting down
producer.Flush(15 * 1000)
producer.Close()
}
func GetInput() []byte {
reader := bufio.NewReader(os.Stdin)
data, _, _ := reader.ReadLine()
return data
}
The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.
package main
import (
"fmt"
"github.com/confluentinc/confluent-kafka-go/kafka"
"log"
"os"
"os/signal"
"syscall"
)
var (
brokers = "ip1:port1,ip2:port2,ip3:port3"
group = "group-id"
topics = "topic_name"
user = "username"
password = "password"
caFile = "phy_ca.crt" // Obtain the SSL certificate by referring to section "Collecting Connection Information". If Security Protocol is set to SASL_PLAINTEXT, delete this parameter.
)
func main() {
log.Println("Starting a new kafka consumer")
config := &kafka.ConfigMap{
"bootstrap.servers": brokers,
"group.id": group,
"auto.offset.reset": "earliest",
"security.protocol": "SASL_SSL",
"sasl.mechanism": "PLAIN",
"sasl.username": user,
"sasl.password": password,
"ssl.ca.location": caFile, // If Security Protocol is set to SASL_PLAINTEXT, delete this parameter.
"ssl.endpoint.identification.algorithm": "none"
}
consumer, err := kafka.NewConsumer(config)
if err != nil {
log.Panicf("Error creating consumer: %v", err)
return
}
err = consumer.SubscribeTopics([]string{topics}, nil)
if err != nil {
log.Panicf("Error subscribe consumer: %v", err)
return
}
go func() {
for {
msg, err := consumer.ReadMessage(-1)
if err != nil {
log.Printf("Consumer error: %v (%v)", err, msg)
} else {
fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
}
}
}()
sigterm := make(chan os.Signal, 1)
signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
select {
case <-sigterm:
log.Println("terminating: via signal")
}
if err = consumer.Close(); err != nil {
log.Panicf("Error closing consumer: %v", err)
}
}
The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.
package main
import (
"fmt"
"github.com/confluentinc/confluent-kafka-go/kafka"
"log"
"os"
"os/signal"
"syscall"
)
var (
brokers = "ip1:port1,ip2:port2,ip3:port3"
group = "group-id"
topics = "topic_name"
)
func main() {
log.Println("Starting a new kafka consumer")
config := &kafka.ConfigMap{
"bootstrap.servers": brokers,
"group.id": group,
"auto.offset.reset": "earliest",
}
consumer, err := kafka.NewConsumer(config)
if err != nil {
log.Panicf("Error creating consumer: %v", err)
return
}
err = consumer.SubscribeTopics([]string{topics}, nil)
if err != nil {
log.Panicf("Error subscribe consumer: %v", err)
return
}
go func() {
for {
msg, err := consumer.ReadMessage(-1)
if err != nil {
log.Printf("Consumer error: %v (%v)", err, msg)
} else {
fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
}
}
}()
sigterm := make(chan os.Signal, 1)
signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
select {
case <-sigterm:
log.Println("terminating: via signal")
}
if err = consumer.Close(); err != nil {
log.Panicf("Error closing consumer: %v", err)
}
}
The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.