~~~ package kafka import ( "github.com/Shopify/sarama" "sync" "project/library/dao/common" "project/library/dao/config" ) type KafkaMgr struct { Conn *sarama.Consumer Wg *sync.WaitGroup PdConn *sarama.SyncProducer } func InitConsumer(cfg *config.KafkaCfg) *KafkaMgr { conn, err := sarama.NewConsumer(cfg.Addrs, nil) if err != nil { common.SimplePanic("InitConsumer失敗", err.Error()) } return &KafkaMgr{Conn: &conn} } func InitProducer(cfg *config.KafkaCfg) *KafkaMgr { config := sarama.NewConfig() config.Producer.RequiredAcks = sarama.WaitForAll //等待服務器所有副本都保存成功后的響應 config.Producer.Partitioner = sarama.NewRandomPartitioner //隨機的分區類型 config.Producer.Return.Successes = true //是否等待成功和失敗后的響應,只有上面的RequireAcks設置不是NoReponse這里才有用. conn, err := sarama.NewSyncProducer(cfg.Addrs, config) if err != nil { common.SimplePanic("InitProducer失敗", err.Error()) } return &KafkaMgr{PdConn: &conn} } ~~~