kafka

Documentation

Index

• type Base
• type Consumer
  • func NewConsumer(addrs []string, topic string, version string, f func(b []byte) error) *Consumer
  • func (p *Consumer) Consume(ctx context.Context) (topErr error)
  • func (p *Consumer) Init() (topErr error)
• type GroupHandler
  • func (c *GroupHandler) Cleanup(ss sarama.ConsumerGroupSession) error
  • func (c *GroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) (topErr error)
  • func (c *GroupHandler) Setup(ss sarama.ConsumerGroupSession) error
• type Producer
  • func NewProducer(addrs []string, topic string) *Producer
  • func (p *Producer) SyncClose() (topErr error)
  • func (p *Producer) SyncInit() (topErr error)
  • func (p *Producer) SyncSend(data []byte) (partition int32, offset int64, topErr error)

Types

type Base

type Base struct {
	Config *sarama.Config
	// contains filtered or unexported fields
}

type Consumer

type Consumer struct {
	*Base

	Func   func(b []byte) error //接收json的处理方式
	Client sarama.ConsumerGroup
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(addrs []string, topic string, version string, f func(b []byte) error) *Consumer

NewConsumer 创建消费者 创建消费者后可以调整config相关参数 addrs地址数组["172.0.0.1:9092","172.0.0.1:9093"] version kafka版本号 f 处理方法

func (*Consumer) Consume

func (p *Consumer) Consume(ctx context.Context) (topErr error)

func (*Consumer) Init

func (p *Consumer) Init() (topErr error)

type GroupHandler

type GroupHandler struct {
	Func func(b []byte) error
}

GroupHandler represents a Sarama consumer group consumer 在分区变更时会触发GroupHandler退出重启

func (*GroupHandler) Cleanup

func (c *GroupHandler) Cleanup(ss sarama.ConsumerGroupSession) error

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

func (*GroupHandler) ConsumeClaim

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

func (*GroupHandler) Setup

func (c *GroupHandler) Setup(ss sarama.ConsumerGroupSession) error

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

type Producer

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

func NewProducer

func NewProducer(addrs []string, topic string) *Producer

NewProducer 创建生产者后需要调用init初始化，初始化之前可以调整config相关参数

func (*Producer) SyncClose

func (p *Producer) SyncClose() (topErr error)

SyncClose 关闭方法在系统退出时可以调用

func (*Producer) SyncInit

func (p *Producer) SyncInit() (topErr error)

SyncInit 初始化同步模式生产者

func (*Producer) SyncSend

func (p *Producer) SyncSend(data []byte) (partition int32, offset int64, topErr error)