You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
ntool-biz/core/mq/kafka/producer.go

71 lines
1.8 KiB
Go

1 year ago
package kafka
import (
"git.noahlan.cn/noahlan/ntool/ndef"
"github.com/Shopify/sarama"
)
type Producer struct {
topic string
client sarama.Client
producer sarama.AsyncProducer
marshaler ndef.Marshaler
}
type ProducerConfig struct {
RequiredAcks sarama.RequiredAcks
Partitioner sarama.PartitionerConstructor
IsReturnSuccess bool
Marshaler ndef.Marshaler
}
func NewKafkaProducer(cfg *ProducerConfig, addr []string, topic string) *Producer {
p := Producer{}
config := sarama.NewConfig() // Instantiate a sarama Config
config.Producer.Return.Successes = cfg.IsReturnSuccess // Whether to enable the successes channel to be notified after the message is sent successfully
config.Producer.RequiredAcks = cfg.RequiredAcks // Set producer Message Reply level 0 1 all
config.Producer.Partitioner = cfg.Partitioner // Set the hash-key automatic hash partition. When sending a message, you must specify the key value of the message. If there is no key, the partition will be selected randomly
p.marshaler = cfg.Marshaler
p.topic = topic
var err error
p.client, err = sarama.NewClient(addr, config)
if err != nil {
return &p
}
p.producer, err = sarama.NewAsyncProducerFromClient(p.client)
if err != nil {
return &p
}
go func() {
for range p.producer.Successes() {
}
}()
return &p
}
func (p *Producer) SendMessageAsync(v interface{}, key ...string) error {
kMsg := &sarama.ProducerMessage{}
kMsg.Topic = p.topic
if len(key) > 0 {
kMsg.Key = sarama.StringEncoder(key[0])
}
if str, ok := v.(string); ok {
kMsg.Value = sarama.StringEncoder(str)
} else if p.marshaler != nil {
marshal, err := p.marshaler.Marshal(v)
if err != nil {
return err
}
kMsg.Value = sarama.ByteEncoder(marshal)
}
select {
case p.producer.Input() <- kMsg:
}
return nil
}