package kafka
var (
	kafkaClient *Client
func InitKafka() {
	var err error
	var config = Config{
		Host: []string{"kafka:9092"},
	kafkaClient, err = NewClient(config)
	if err != nil {
		panic(err)
func GetClient() *Client {
	return kafkaClient
Producer.go
package kafka
import (
   "errors"
   "github.com/Shopify/sarama"
type Client struct {
   sarama.AsyncProducer
   msgPool chan *sarama.ProducerMessage
type Config struct {
   Host          []string `json:"host"`
   ReturnSuccess bool     `json:"return_success"`
   ReturnErrors  bool     `json:"return_errors"`
func NewClient(cfg Config) (*Client, error) {
   // create client
   var err error
   c := &Client{
      msgPool: make(chan *sarama.ProducerMessage, 2000),
   config := sarama.NewConfig()
   config.Producer.Return.Errors = cfg.ReturnErrors
   config.Producer.Return.Successes = cfg.ReturnSuccess
   config.Version = sarama.V2_0_0_0
   c.AsyncProducer, err = sarama.NewAsyncProducer(cfg.Host, config)
   if err != nil {
      return nil, err
   return c, nil
// run
func (c *Client) Run() {
   for {
      select {
      case msg := <-c.msgPool:
         c.Input() <- msg
         logger.Info("%+v", msg)
// send msg
func (c *Client) Send(topic string, msg []byte) error {
   if topic == "" {
      return errors.New("kafka producer send msg topic empty")
   kafkaMsg := &sarama.ProducerMessage{
      Topic: topic,
      Value: sarama.ByteEncoder(msg),
   c.msgPool <- kafkaMsg
   return nil
生产者初始化：
// kafka init
kafka.InitKafka()
go kafka.GetClient().Run()
consumer.go
package kafka_consumer
import (
   "context"
   "github.com/Shopify/sarama"
   "os/signal"
   "sync"
   "syscall"
// Consumer represents a Sarama consumer group consumer
type Consumer struct {
   ready chan bool
func (c *Consumer) Setup(session sarama.ConsumerGroupSession) error {
   //panic("implement me")
   return nil
func (c *Consumer) Cleanup(session sarama.ConsumerGroupSession) error {
   //panic("implement me")
   return nil
func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
   for message := range claim.Messages() {
      logger.Info("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
      session.MarkMessage(message, "")
      c.Handler(message.Topic, message.Value)
   return nil
func (c *Consumer) Handler(topic string, msg []byte) {
   switch topic {
   case conscom.KafkaTopicGiftOrder:
      GiftOrder(topic, msg)
   case conscom.KafkaTopicFollow:
      UserFollow(topic, msg)
func ConsumeInit(topics []string, groupID string) {
   consumer := Consumer{
      ready: make(chan bool),
   brokerList := []string{"kafka:9092"}
   config := sarama.NewConfig()
   config.Version = sarama.V1_0_0_0
   ctx, cancel := context.WithCancel(context.Background())
   client, err := sarama.NewConsumerGroup(brokerList, groupID, config)
   if err != nil {
      log.Printf("kafka consumer err %v", err)
      return
   wg := &sync.WaitGroup{}
   wg.Add(1)
   go func() {
      defer wg.Done()
      for {
         // server-side rebalance happens, the consumer session will need to be
         if err := client.Consume(ctx, topics, &consumer); err != nil {
            log.Printf("kafka consumer: %v", err)
         // check if context was cancelled, signaling that the consumer should stop
         if ctx.Err() != nil {
            return
         consumer.ready = make(chan bool)
   sigterm := make(chan os.Signal, 1)
   signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
   select {
   case <-ctx.Done():
      log.Printf("kafka consume gift terminating: context cancelled")
   case <-sigterm:
      log.Printf("kafka consume gift terminating: via signal")
   cancel()
   wg.Wait()
   if err = client.Close(); err != nil {
      log.Printf("kafka consume gift Error closing client: %v", err)
消费者初始化：
// kafka consumer
go kafka_consumer.ConsumeInit([]string{"topicA", "topicB", "group-name")
《深入理解Kafka:核心设计与实践原理》作者:朱忠华
https://github.com/Shopify/sarama
http://kafka.apache.org/documentation/
https://crossoverjie.top/2018/11/20/kafka/kafka-consumer/