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.
71 lines
1.8 KiB
Go
71 lines
1.8 KiB
Go
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
|
|
}
|