package KafkaUtil

import (
	"dsSzxy/Utils/ConfigUtil"
	"fmt"
	"github.com/Shopify/sarama"
)

func SendKafkaMsg(topic string, content string) error {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true

	producer, err := sarama.NewAsyncProducer([]string{ConfigUtil.KafkaAddrs}, config)
	if err != nil {
		return err
	}
	defer producer.AsyncClose()
	msg := &sarama.ProducerMessage{
		Topic: topic,
	}
	msg.Value = sarama.ByteEncoder(content)

	producer.Input() <- msg
	select {
	case suc := <-producer.Successes():
		fmt.Printf(suc.Topic + "Topic发送成功！")
	case fail := <-producer.Errors():
		return fail.Err
	}
	return nil
}