在Golang中操作Kafka时，处理错误的关键是检查每个操作返回的错误值。以下是一些常见的Kafka操作及其错误处理方法：

创建生产者：

package mainimport ("fmt""github.com/Shopify/sarama")func main() {config := sarama.NewConfig()config.Producer.RequiredAcks = sarama.WaitForAllconfig.Producer.Retry.Max = 5config.Producer.Return.Successes = trueproducer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)if err != nil {fmt.Printf("Error creating producer: %v\n", err)return}defer func() {if err := producer.Close(); err != nil {fmt.Printf("Error closing producer: %v\n", err)}}()// 生产者创建成功，可以进行后续操作}

发送消息：

msg := &sarama.ProducerMessage{Topic: "test_topic",Value: sarama.StringEncoder("Hello, Kafka!"),}partition, offset, err := producer.SendMessage(msg)if err != nil {fmt.Printf("Error sending message: %v\n", err)return}fmt.Printf("Message sent to partition %d at offset %d\n", partition, offset)

消费消息：

consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)if err != nil {fmt.Printf("Error creating consumer: %v\n", err)return}defer func() {if err := consumer.Close(); err != nil {fmt.Printf("Error closing consumer: %v\n", err)}}()partitionConsumer, err := consumer.ConsumePartition("test_topic", 0, sarama.OffsetNewest)if err != nil {fmt.Printf("Error consuming partition: %v\n", err)return}defer func() {if err := partitionConsumer.Close(); err != nil {fmt.Printf("Error closing partition consumer: %v\n", err)}}()for msg := range partitionConsumer.Messages() {fmt.Printf("Received message: %s (partition %d, offset %d)\n", string(msg.Value), msg.Partition, msg.Offset)}

在上述示例中，我们使用了sarama库来操作Kafka。在每个操作中，我们都检查了返回的错误值，并在发生错误时进行了相应的处理。这样可以确保我们在遇到问题时能够及时发现并采取相应的措施。