kafka

package
v0.0.0-...-440800f Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 27, 2023 License: Apache-2.0 Imports: 6 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ConsumerGroup

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

ConsumerGroup kafka consumer

func MustKafkaConsumer

func MustKafkaConsumer(c *KafkaConsumerConf) *ConsumerGroup

func (*ConsumerGroup) Cleanup

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*ConsumerGroup) ConsumeClaim

func (c *ConsumerGroup) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*ConsumerGroup) Group

func (c *ConsumerGroup) Group() string

func (*ConsumerGroup) RegisterHandlers

func (c *ConsumerGroup) RegisterHandlers(topic string, cb MessageHandlerF)

func (*ConsumerGroup) Setup

Setup is run at the beginning of a new session, before ConsumeClaim

func (*ConsumerGroup) Start

func (c *ConsumerGroup) Start()

Start start consume messages, watch signals

func (*ConsumerGroup) Stop

func (c *ConsumerGroup) Stop()

Stop Stop consume messages, watch signals

func (*ConsumerGroup) Topics

func (c *ConsumerGroup) Topics() []string

type KafkaConsumerConf

type KafkaConsumerConf struct {
	Brokers []string
	Topics  []string
	Group   string
}

KafkaConsumerConf kafka client settings.

type KafkaProducerConf

type KafkaProducerConf struct {
	Brokers []string
	Topic   string
}

KafkaProducerConf kafka producer settings.

type MessageHandlerF

type MessageHandlerF func(ctx context.Context, key string, value []byte)

type Producer

type Producer struct {
	SyncProducer sarama.SyncProducer
	// contains filtered or unexported fields
}

func GetCachedMQClient

func GetCachedMQClient(c *KafkaProducerConf) *Producer

func MustKafkaProducer

func MustKafkaProducer(c *KafkaProducerConf) *Producer

func (*Producer) Close

func (p *Producer) Close() (err error)

func (*Producer) SendMessage

func (p *Producer) SendMessage(ctx context.Context, key string, value []byte) (int32, int64, error)

SendMessage Input send msg to kafka NOTE: If producer has beed created failed, the message will lose.

func (*Producer) Topic

func (p *Producer) Topic() string

Directories

Path Synopsis
examples

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL