kafka

Documentation

Index

• type HandCommitConsumer
  • func (consumer *HandCommitConsumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (consumer *HandCommitConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *HandCommitConsumer) Setup(sarama.ConsumerGroupSession) error
• type KafkaMessageQueueConsumer
  • func NewDefaultKafkaMessageQueueConsumer(brokers []string, groupName string, autoCommit bool, ctx context.Context) (*KafkaMessageQueueConsumer, error)
  • func (c *KafkaMessageQueueConsumer) Close() error
  • func (c *KafkaMessageQueueConsumer) ConsumeWithHandCommit(topics []string, handleMsg func(msg *sarama.ConsumerMessage) bool)
• type KafkaMessageQueueProducer
  • func NewDefaultKafkaMessageQueueProducer(brokersAddress []string, reqTryCfg RetryConfig) (*KafkaMessageQueueProducer, error)
  • func (k *KafkaMessageQueueProducer) Close()
  • func (k *KafkaMessageQueueProducer) CloseKafkaProducer()
  • func (k *KafkaMessageQueueProducer) SendJsonMessage(topic string, jsonByte []byte) error
  • func (k *KafkaMessageQueueProducer) SendJsonMessages(topic string, jsonBytes ...[]byte) error
• type RetryConfig

Types

type HandCommitConsumer

type HandCommitConsumer struct {
	HandleMsg func(msg *sarama.ConsumerMessage) bool // true,  msg means consume
}

func (*HandCommitConsumer) Cleanup

func (consumer *HandCommitConsumer) Cleanup(sarama.ConsumerGroupSession) error

func (*HandCommitConsumer) ConsumeClaim

func (consumer *HandCommitConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*HandCommitConsumer) Setup

func (consumer *HandCommitConsumer) Setup(sarama.ConsumerGroupSession) error

type KafkaMessageQueueConsumer

type KafkaMessageQueueConsumer struct {
	ConsumerGroup sarama.ConsumerGroup
	Ctx           context.Context
	// contains filtered or unexported fields
}

func NewDefaultKafkaMessageQueueConsumer

func NewDefaultKafkaMessageQueueConsumer(brokers []string, groupName string, autoCommit bool, ctx context.Context) (*KafkaMessageQueueConsumer, error)

func (*KafkaMessageQueueConsumer) Close

func (c *KafkaMessageQueueConsumer) Close() error

func (*KafkaMessageQueueConsumer) ConsumeWithHandCommit

func (c *KafkaMessageQueueConsumer) ConsumeWithHandCommit(topics []string, handleMsg func(msg *sarama.ConsumerMessage) bool)

handleMsg return true means will commit this msg after handle it

type KafkaMessageQueueProducer

type KafkaMessageQueueProducer struct {
	SyncProducer *sarama.SyncProducer
	ReqTryCfg    RetryConfig
}

func NewDefaultKafkaMessageQueueProducer

func NewDefaultKafkaMessageQueueProducer(brokersAddress []string, reqTryCfg RetryConfig) (*KafkaMessageQueueProducer, error)

func (*KafkaMessageQueueProducer) Close

func (k *KafkaMessageQueueProducer) Close()

func (*KafkaMessageQueueProducer) CloseKafkaProducer

func (k *KafkaMessageQueueProducer) CloseKafkaProducer()

func (*KafkaMessageQueueProducer) SendJsonMessage

func (k *KafkaMessageQueueProducer) SendJsonMessage(topic string, jsonByte []byte) error

single send

func (*KafkaMessageQueueProducer) SendJsonMessages

func (k *KafkaMessageQueueProducer) SendJsonMessages(topic string, jsonBytes ...[]byte) error

batch send

type RetryConfig

type RetryConfig struct {
	RetryTime int
	DelayTime time.Duration
}