kafka

Documentation

Index

• func NewConsumerGroup(nodes []*conf.Kafka_Consumer)
• func NewConsumerWorker(nodes []*conf.Kafka_Consumer) error
• func Register(name string, maker Handler)
• func RegisterConsumer(nodes []*conf.Kafka_Consumer) error
• func RegisterProducer(nodes []*conf.Kafka_Producer) error
• type Consumer
  • func (Consumer) Cleanup(_ sarama.ConsumerGroupSession) error
  • func (c Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (Consumer) Setup(_ sarama.ConsumerGroupSession) error
• type Handler
  • func HandlerInterface() Handler
• type NullHandler
  • func (h *NullHandler) Do(ctx context.Context, msg *sarama.ConsumerMessage) (err error)
• type Producer
  • func NewProducer(topic string) (*Producer, error)
  • func (p *Producer) Send(data string) error
• type WorkerHandler
  • func (n *WorkerHandler) Run(ctx context.Context, msg *sarama.ConsumerMessage) (err error)
  • func (n *WorkerHandler) SetNext(h Handler) Handler

Functions

func NewConsumerGroup

func NewConsumerGroup(nodes []*conf.Kafka_Consumer)

func NewConsumerWorker

func NewConsumerWorker(nodes []*conf.Kafka_Consumer) error

NewConsumerWorker yaml配置topic对于的func调用链式实例

func Register

func Register(name string, maker Handler)

func RegisterConsumer

func RegisterConsumer(nodes []*conf.Kafka_Consumer) error

func RegisterProducer

func RegisterProducer(nodes []*conf.Kafka_Producer) error

Types

type Consumer

type Consumer struct{}

func (Consumer) Cleanup

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

func (Consumer) ConsumeClaim

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

func (Consumer) Setup

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

type Handler

type Handler interface {
	Do(ctx context.Context, msg *sarama.ConsumerMessage) error
	SetNext(h Handler) Handler
	Run(ctx context.Context, msg *sarama.ConsumerMessage) error
}

func HandlerInterface

func HandlerInterface() Handler

type NullHandler

type NullHandler struct {
	WorkerHandler
}

首链

func (*NullHandler) Do

func (h *NullHandler) Do(ctx context.Context, msg *sarama.ConsumerMessage) (err error)

type Producer

type Producer struct {
	Topic        string              `json:"topic"`
	SyncProducer sarama.SyncProducer `json:"map"`
}

func NewProducer

func NewProducer(topic string) (*Producer, error)

func (*Producer) Send

func (p *Producer) Send(data string) error

type WorkerHandler

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

func (*WorkerHandler) Run

func (n *WorkerHandler) Run(ctx context.Context, msg *sarama.ConsumerMessage) (err error)

func (*WorkerHandler) SetNext

func (n *WorkerHandler) SetNext(h Handler) Handler