ckafka

Documentation

Index

• type ConsumerGroup
  • func GetConsumer(cfg ConsumerGroupConfig) *ConsumerGroup
  • func (cg *ConsumerGroup) RegisterHandleAndConsumer(handler msgConsumer)
• type ConsumerGroupConfig
  • func (c ConsumerGroupConfig) GetOffset() kafka.ConfigValue
• type Producer
  • func MustNewProducer(config ProducerConfig) *Producer
  • func (p *Producer) SendMessage(ctx context.Context, bMsg []byte, key string) (partition int32, offset int64, err error)
• type ProducerConfig

Types

type ConsumerGroup

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

func GetConsumer

func GetConsumer(cfg ConsumerGroupConfig) *ConsumerGroup

func (*ConsumerGroup) RegisterHandleAndConsumer

func (cg *ConsumerGroup) RegisterHandleAndConsumer(handler msgConsumer)

type ConsumerGroupConfig

type ConsumerGroupConfig struct {
	Brokers   []string
	Topic     string
	Group     string
	Offset    string `json:",options=first|last,default=last"`
	Consumers int    `json:",default=8"`
	User      string `json:",optional"`
	Passwd    string `json:",optional"`
}

func (ConsumerGroupConfig) GetOffset

func (c ConsumerGroupConfig) GetOffset() kafka.ConfigValue

type Producer

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

func MustNewProducer

func MustNewProducer(config ProducerConfig) *Producer

func (*Producer) SendMessage

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

type ProducerConfig

type ProducerConfig struct {
	Brokers   []string `json:""`
	Topic     string   `json:""`
	User      string   `json:",optional"`
	Passwd    string   `json:",optional"`
	Partition int      `json:",default=1"`
}