初始化的主要代码如下：

    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll // Wait for all in-sync replicas to ack the message
    config.Producer.Retry.Max = 10                   // Retry up to 10 times to produce the message
    config.Producer.Return.Successes = true
    config.Producer.Compression = sarama.CompressionSnappy // Compress messages
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    config.Producer.Return.Errors = true
    producer, err := sarama.NewSyncProducer(brokerList, config)
    if err != nil {
        log.Printf("sarama.NewSyncProducer fail:%+v\n", err)
        return err
    }

生产消息的主要代码如下：

    msg := sarama.ProducerMessage{
        Topic: monitorInstance.topic,
        Value: sarama.StringEncoder(resportString),
        //Key: sarama.StringEncoder("monitor_data"),
    }
    partition, offset, err := (*producer).SendMessage(&msg)
    if err != nil {
        log.Printf("kafka SendMessage fail:%+v\n", err)
        return
    }
    log.Printf("Your data is stored with unique identifier important/%d/%d\n", partition, offset)

尝试了以下办法都不行：

1.把producer从async修改为sync

2.增加msg中的key

3.尝试更换topic

最后发现是这个问题：

config.Producer.Compression = sarama.CompressionSnappy

注释掉这一行后正常。

难道就不能加压缩？问了腾讯云的工程师，得到了解决：

config.Version = sarama.V2_1_0_0

config.Producer.Compression = sarama.CompressionSnappy