mv src/common src/utils src/grpc-etcdv3 to pkg

pkg/common/kafka/consumer.go

@@ -0,0 +1,36 @@
+package kafka
+
+import (
+	"github.com/Shopify/sarama"
+	"sync"
+)
+
+type Consumer struct {
+	addr          []string
+	WG            sync.WaitGroup
+	Topic         string
+	PartitionList []int32
+	Consumer      sarama.Consumer
+}
+
+func NewKafkaConsumer(addr []string, topic string) *Consumer {
+	p := Consumer{}
+	p.Topic = topic
+	p.addr = addr
+
+	consumer, err := sarama.NewConsumer(p.addr, nil)
+	if err != nil {
+		panic(err)
+		return nil
+	}
+	p.Consumer = consumer
+
+	partitionList, err := consumer.Partitions(p.Topic)
+	if err != nil {
+		panic(err)
+		return nil
+	}
+	p.PartitionList = partitionList
+
+	return &p
+}

pkg/common/kafka/consumer_group.go

@@ -0,0 +1,53 @@
+/*
+** description("").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/5/11 9:36).
+ */
+package kafka
+
+import (
+	"context"
+	"github.com/Shopify/sarama"
+)
+
+type MConsumerGroup struct {
+	sarama.ConsumerGroup
+	groupID string
+	topics  []string
+}
+
+type MConsumerGroupConfig struct {
+	KafkaVersion   sarama.KafkaVersion
+	OffsetsInitial int64
+	IsReturnErr    bool
+}
+
+func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addr []string, groupID string) *MConsumerGroup {
+	config := sarama.NewConfig()
+	config.Version = consumerConfig.KafkaVersion
+	config.Consumer.Offsets.Initial = consumerConfig.OffsetsInitial
+	config.Consumer.Return.Errors = consumerConfig.IsReturnErr
+	client, err := sarama.NewClient(addr, config)
+	if err != nil {
+		panic(err)
+	}
+	consumerGroup, err := sarama.NewConsumerGroupFromClient(groupID, client)
+	if err != nil {
+		panic(err)
+	}
+	return &MConsumerGroup{
+		consumerGroup,
+		groupID,
+		topics,
+	}
+}
+func (mc *MConsumerGroup) RegisterHandleAndConsumer(handler sarama.ConsumerGroupHandler) {
+	ctx := context.Background()
+	for {
+		err := mc.ConsumerGroup.Consume(ctx, mc.topics, handler)
+		if err != nil {
+			panic(err)
+		}
+	}
+}

pkg/common/kafka/producer.go

@@ -0,0 +1,49 @@
+package kafka
+
+import (
+	log2 "Open_IM/pkg/common/log"
+	"github.com/Shopify/sarama"
+	"github.com/golang/protobuf/proto"
+)
+
+type Producer struct {
+	topic    string
+	addr     []string
+	config   *sarama.Config
+	producer sarama.SyncProducer
+}
+
+func NewKafkaProducer(addr []string, topic string) *Producer {
+	p := Producer{}
+	p.config = sarama.NewConfig()                             //Instantiate a sarama Config
+	p.config.Producer.Return.Successes = true                 //Whether to enable the successes channel to be notified after the message is sent successfully
+	p.config.Producer.RequiredAcks = sarama.WaitForAll        //Set producer Message Reply level 0 1 all
+	p.config.Producer.Partitioner = sarama.NewHashPartitioner //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.addr = addr
+	p.topic = topic
+
+	producer, err := sarama.NewSyncProducer(p.addr, p.config) //Initialize the client
+	if err != nil {
+		panic(err)
+		return nil
+	}
+	p.producer = producer
+	return &p
+}
+
+func (p *Producer) SendMessage(m proto.Message, key ...string) (int32, int64, error) {
+	kMsg := &sarama.ProducerMessage{}
+	kMsg.Topic = p.topic
+	if len(key) == 1 {
+		kMsg.Key = sarama.StringEncoder(key[0])
+	}
+	bMsg, err := proto.Marshal(m)
+	if err != nil {
+		log2.Error("", "", "proto marshal err = %s", err.Error())
+		return -1, -1, err
+	}
+	kMsg.Value = sarama.ByteEncoder(bMsg)
+
+	return p.producer.SendMessage(kMsg)
+}