gokafka

Documentation

Index

• Variables
• func Close()
• func NewConsumerConfig(brokers []string, groupID string, topic string) kafka.ReaderConfig
• func NewProducerConfig(brokers []string) *kafka.Writer
• func NewVipConsumerConfig(brokername string, groupID string, topic string) kafka.ReaderConfig
• func NewVipProducerConfig(configName string) *kafka.Writer
• type Consumer
  • func NewConsumer(conf kafka.ReaderConfig) *Consumer
  • func (kr *Consumer) Close() error
  • func (kr *Consumer) Handle(ctx context.Context, handle func(msg kafka.Message) error) error
• type Producer
  • func GetProducer(configName string) (conn *Producer)
  • func NewProducer(configName string, conf *kafka.Writer) *Producer
  • func (w *Producer) Send(ctx context.Context, topic string, key string, value []byte) error
  • func (w *Producer) SendBatch(ctx context.Context, msgs ...kafka.Message) error

Variables

var ProducerPool sync.Map

Functions

func Close

func Close()

func NewConsumerConfig

func NewConsumerConfig(brokers []string, groupID string, topic string) kafka.ReaderConfig

func NewProducerConfig

func NewProducerConfig(brokers []string) *kafka.Writer

NewProducerConfig 该方法返回值不能复用，每次NewProducer时都需要调用一次

func NewVipConsumerConfig

func NewVipConsumerConfig(brokername string, groupID string, topic string) kafka.ReaderConfig

func NewVipProducerConfig

func NewVipProducerConfig(configName string) *kafka.Writer

NewVipProducerConfig 该方法返回值不能复用，每次NewProducer时都需要调用一次

Types

type Consumer

type Consumer struct {
	Reader *kafka.Reader
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(conf kafka.ReaderConfig) *Consumer

NewConsumer conf每次重新生成

func (*Consumer) Close

func (kr *Consumer) Close() error

func (*Consumer) Handle

func (kr *Consumer) Handle(ctx context.Context, handle func(msg kafka.Message) error) error

type Producer

type Producer struct {
	Writer *kafka.Writer
	// contains filtered or unexported fields
}

func GetProducer

func GetProducer(configName string) (conn *Producer)

func NewProducer

func NewProducer(configName string, conf *kafka.Writer) *Producer

NewProducer conf每次重新生成

func (*Producer) Send

func (w *Producer) Send(ctx context.Context, topic string, key string, value []byte) error

func (*Producer) SendBatch

func (w *Producer) SendBatch(ctx context.Context, msgs ...kafka.Message) error