golang使用sarama生产和消费kafka

蔚落 2022-12-15 12:40 309阅读 0赞

关于kafka相关架构及应用场景等，请见笔者[相关文章][Link 1]。

## 1.消费kafka消息 ##

package main
    
    import (
    	"fmt"
    	"github.com/Shopify/sarama"
    	cluster "github.com/bsm/sarama-cluster"
    	"time"
    )
    
    
    var (
    	kafkaConsumer *cluster.Consumer
    	kafkaBrokers = []string{ "127.0.0.1:9092"}
    	kafkaTopic   = "test_topic_1"
    	groupId = "csdn_test_1"
    )
    
    func init() { 
    	var err error
    	config := cluster.NewConfig()
    	config.Consumer.Return.Errors = true
    	config.Group.Return.Notifications = true
    	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
    	config.Consumer.Offsets.Initial = -2
    	config.Consumer.Offsets.CommitInterval = 1 * time.Second
    	config.Group.Return.Notifications = true
    	kafkaConsumer, err = cluster.NewConsumer(kafkaBrokers, groupId, []string{ kafkaTopic}, config)
    	if err != nil { 
    		panic(err.Error())
    	}
    	if kafkaConsumer == nil { 
    		panic(fmt.Sprintf("consumer is nil. kafka info -> {brokers:%v, topic: %v, group: %v}", kafkaBrokers, kafkaTopic, groupId))
    	}
    	fmt.Printf("kafka init success, consumer -> %v, topic -> %v, ", kafkaConsumer, kafkaTopic)
    }
    
    func main() { 
    	for { 
    		select { 
    		case msg, ok := <-kafkaConsumer.Messages():
    			if ok { 
    				fmt.Printf("kafka msg: %s \n", msg.Value)
    				kafkaConsumer.MarkOffset(msg, "")
    			} else { 
    				fmt.Printf("kafka 监听服务失败")
    			}
    		case err, ok := <-kafkaConsumer.Errors():
    			if ok { 
    				fmt.Printf("consumer error: %v" , err)
    			}
    		case ntf, ok := <-kafkaConsumer.Notifications():
    			if ok { 
    				fmt.Printf("consumer notification: %v" , ntf)
    			}
    		}
    	}
    }

## 2.生产kafka消息 ##

package main
    
    import (
    	"bytes"
    	"encoding/json"
    	"fmt"
    	"github.com/Shopify/sarama"
    	"time"
    )
    
    func main() { 
    	startProduce()
    }
    
    var (
    	producer sarama.SyncProducer
    	brokers = []string{ "127.0.0.1:9092"}
    	topic   = "test_topic_1"
    )
    
    func init() { 
    	config := sarama.NewConfig()
    	config.Producer.RequiredAcks = sarama.WaitForLocal
    	config.Producer.Retry.Max = 5
    	config.Producer.Return.Successes = true
    	brokers := brokers
    	var err error
    	producer, err = sarama.NewSyncProducer(brokers, config)
    	if err != nil { 
    		fmt.Printf("init producer failed -> %v \n", err)
    		panic(err)
    	}
    	fmt.Println("producer init success")
    }
    
    func produceMsg(msg string) { 
    	msgX := &sarama.ProducerMessage{ 
    		Topic: topic,
    		Value: sarama.StringEncoder(msg),
    	}
    	fmt.Printf("SendMsg -> %v\n", dumpString(msgX))
    
    	partition, offset, err := producer.SendMessage(msgX)
    	if err != nil { 
    		fmt.Printf("send msg error:%s \n", err)
    	}
    	fmt.Printf("msg send success, message is stored in topic(%s)/partition(%d)/offset(%d)\n", topic, partition, offset)
    }
    
    func startProduce() { 
    	tick := time.Tick(2 * time.Second)
    	for { 
    		select { 
    		case <-tick:
    			t := time.Now().Unix() * 1000
    			msg := fmt.Sprintf("{\"timestamp\":%d}", t)
    			produceMsg(msg)
    		}
    	}
    }
    
    //解析为json字符串
    func dumpString(v interface{ }) (str string) { 
    
    	bs, err := json.Marshal(v)
    	b := bytes.Buffer{ }
    	if err != nil { 
    		b.WriteString("{err:\"json format error.")
    		b.WriteString(err.Error())
    		b.WriteString("\"}")
    	} else { 
    		b.Write(bs)
    	}
    	str = b.String()
    	return str
    }

[Link 1]: https://blog.csdn.net/m0_38075425/article/details/81353738