How to use Golang and Kafka together

Kafka is an open source distributed message queue that is often used to build real-time data stream processing applications in big data applications. Golang is a programming language developed by Google and is known for its efficient concurrency, powerful libraries and ecosystem. So, how to use Golang to combine with Kafka?

First, we need to import the github.com/Shopify/sarama package. This is a Golang client library that supports Kafka. During the installation process, you need to run the following command:

go get github.com/Shopify/sarama

Next, we need to create a producer. First, create the configuration:

config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Retry.Max = 5
config.Producer.Return.Successes = true

Here we set the producer to wait for all ACKs, try up to 5 retries, and return a success message to the producer after success.

Next, we need to create a producer instance:

producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
if err != nil {
    panic(err)
}
defer producer.Close()

We need to specify a Kafka broker address as the service endpoint to connect to Kafka. Here we are connecting to the local Kafka server. We also call the .Close() method to ensure cleanup when the producer exits.

Now we are ready to start publishing messages to the Kafka topic:

msg := &sarama.ProducerMessage{
    Topic: "test",
    Value: sarama.StringEncoder("Hello World!"),
}

part, offset, err := producer.SendMessage(msg)
if err != nil {
    fmt.Printf("Error publishing message: %v", err)
} else {
    fmt.Printf("Message published successfully. Partition: %v, Offset: %v\n", part, offset)
}

In this example, we publish a message to the topic named "test". If there are no errors, it prints out the successfully published partition and offset.

Now we have created a producer that publishes a message to Kafka. Next, let's take a look at how to create a consumer.

First, we need to create the consumer configuration:

config := sarama.NewConfig()
config.Consumer.Return.Errors = true

Here we set the receiving error.

Next, we need to create a consumer instance:

consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
if err != nil {
    panic(err)
}
defer consumer.Close()

Here we also specify a Kafka broker address. We also need to call the .Close() method to ensure that the consumer will clean up when it exits.

Now we are ready to read messages from the Kafka topic:

partitionConsumer, err := consumer.ConsumePartition("test", 0, sarama.OffsetOldest)
if err != nil {
    panic(err)
}
defer partitionConsumer.Close()

for {
    select {
    case msg := <-partitionConsumer.Messages():
        fmt.Printf("Received message from partition %d with offset %d: %s = %s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
    case err := <-partitionConsumer.Errors():
        fmt.Println("Error: ", err.Error())
    }
}

In this example, we subscribe to the topic named "test". Then we read the offset of the first partition. We then read messages from that partition infinitely in a loop. The select statement in the loop will always monitor the message and error channels and print them respectively.

So far, we have introduced how to use Golang and Kafka to combine. With this simple example, you should have mastered the basic usage of Golang and Kafka.

The above is the detailed content of How to use Golang and Kafka together. For more information, please follow other related articles on the PHP Chinese website!