kafka

Documentation

Overview

** description(""). ** copyright('tuoyun,www.tuoyun.net'). ** author("fg,Gordon@tuoyun.net"). ** time(2021/5/11 9:36).

Index

• type Consumer
  • func NewKafkaConsumer(addr []string, topic string) *Consumer
• type MConsumerGroup
  • func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []string, groupID string) *MConsumerGroup
  • func (mc *MConsumerGroup) RegisterHandleAndConsumer(handler sarama.ConsumerGroupHandler)
• type MConsumerGroupConfig
• type Producer
  • func NewKafkaProducer(addr []string, topic string) *Producer
  • func (p *Producer) SendMessage(m proto.Message, key string, operationID string) (int32, int64, error)

Types

type Consumer

type Consumer struct {
	WG            sync.WaitGroup
	Topic         string
	PartitionList []int32
	Consumer      sarama.Consumer
	// contains filtered or unexported fields
}

func NewKafkaConsumer

func NewKafkaConsumer(addr []string, topic string) *Consumer

type MConsumerGroup

type MConsumerGroup struct {
	sarama.ConsumerGroup
	// contains filtered or unexported fields
}

func NewMConsumerGroup

func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []string, groupID string) *MConsumerGroup

func (*MConsumerGroup) RegisterHandleAndConsumer

func (mc *MConsumerGroup) RegisterHandleAndConsumer(handler sarama.ConsumerGroupHandler)

type MConsumerGroupConfig

type MConsumerGroupConfig struct {
	KafkaVersion   sarama.KafkaVersion
	OffsetsInitial int64
	IsReturnErr    bool
}

type Producer

type Producer struct {
	// contains filtered or unexported fields
}

func NewKafkaProducer

func NewKafkaProducer(addr []string, topic string) *Producer

func (*Producer) SendMessage

func (p *Producer) SendMessage(m proto.Message, key string, operationID string) (int32, int64, error)