封装一个golang调用kafka的工具包，包含了consumer,producer,auth,在自己的生产环境上做过验证。可以做参考作用，也可以直接使用。
部分代码
// Run 执行消费动作
func (cg *ConsumerGroup) Run(ctx context.Context) {defer cg.close()for {select {case err := <-cg.consumer.Errors():cg.logger.WithError(err).Errorln("Error channel")cg.handleConsumeError(err)case <-ctx.Done():err := ctx.Err()if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {// 正常退出return}cg.logger.WithError(err).Errorln("上下文异常退出")default:if err := cg.consumer.Consume(ctx, cg.options.topics, cg.options); err != nil {cg.logger.WithError(err).Errorln("Consume Error channel")cg.handleConsumeError(err)}}}
}
func (kc *KafkaClient) CreateTopic(ctx context.Context, topicName string, ops ...TopicOption) (err error) {topicConf := &sarama.TopicDetail{NumPartitions: 3,ReplicationFactor: 1,ConfigEntries: map[string]*string{"cleanup.policy": &TopicTTLPolicy,"retention.ms": &TopicTTLRetention,},}for _, op := range ops {op(topicConf)}return kc.cli.CreateTopic(topicName, topicConf, false)
}
代码太多，全部写出来不现实，详细代码在这个下载包里，有test文件可以查看是如何调用的
