kafka

README

Reference

• https://github.com/asong2020/Golang_Dream/tree/master/code_demo/kafka_demo

Documentation

Index

• type Consumer
  • func NewConsumer(config *sarama.Config, logger *log.Logger, topic string, groupID string, ...) *Consumer
  • func (c *Consumer) Consume()
  • func (c Consumer) Stop()
• type ConsumerGroupHandler
  • func (ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error
  • func (h ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error
• type Producer
  • func NewProducer(config *sarama.Config, logger *log.Logger, topic string, brokers []string) *Producer
  • func (p *Producer) Publish(message string)

Types

type Consumer

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

Consumer kafka consumer

func NewConsumer

func NewConsumer(config *sarama.Config, logger *log.Logger, topic string, groupID string, brokers []string, handler *ConsumerGroupHandler) *Consumer

NewConsumer create a consumer nolint

func (*Consumer) Consume

func (c *Consumer) Consume()

Consume consume data

func (Consumer) Stop

func (c Consumer) Stop()

Stop close conn

type ConsumerGroupHandler

type ConsumerGroupHandler struct{}

ConsumerGroupHandler represents the sarama consumer group

func (ConsumerGroupHandler) Cleanup

func (ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error

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

func (ConsumerGroupHandler) ConsumeClaim

func (h ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages(), here is supposed to be what you want to do with the message. In this example the message will be logged with the topic name, partition and message value.

func (ConsumerGroupHandler) Setup

func (ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error

Setup is run before consumer start consuming, is normally used to setup things such as database connections

type Producer

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

Producer kafka producer

func NewProducer

func NewProducer(config *sarama.Config, logger *log.Logger, topic string, brokers []string) *Producer

NewProducer create producer nolint

func (*Producer) Publish

func (p *Producer) Publish(message string)

Publish push data to queue