databus

Documentation

Index

• Constants
• type ConsumerDeal
• type ConsumerEvent
  • func NewConsumer(param ConsumerParam) (ConsumerEvent, error)
• type ConsumerParam
• type ProducerEvent
  • func NewAsyncProducer(param ProducerParam) (ProducerEvent, error)
• type ProducerParam

Constants

const (
	//返回一个手动选择分区的分割器,也就是获取msg中指定的`partition`
	KafkaManual uint32 = 1

	//通过随机函数随机获取一个分区号
	KafkaRandom uint32 = 2

	//环形选择,也就是在所有分区中循环选择一个
	KafkaRoundRobin uint32 = 3

	//通过msg中的key生成hash值,选择分区
	KafkaHash uint32 = 4
)

Types

type ConsumerDeal

type ConsumerDeal interface {
	//数据处理的实现
	DealMessage(data []byte, topic string, partition int32, offset int64, groupid string) error

	//消费组增加消费者的消息通知
	Setup(topicAndPartitions map[string][]int32, memberId string, generationId int32)

	//消费组中消费者退出的消息通知
	Cleanup(topicAndPartitions map[string][]int32, memberId string, generationId int32)
}

使用者必须实现的接口

type ConsumerEvent

type ConsumerEvent interface {
	//启动轮询消费数据
	Start() error

	//关闭消费者，必须调用
	Close()
}

func NewConsumer

func NewConsumer(param ConsumerParam) (ConsumerEvent, error)

type ConsumerParam

type ConsumerParam struct {
	Address   []string
	GroupId   string
	Topic     string
	KafkaVer  string
	Dealhanle ConsumerDeal
}

type ProducerEvent

type ProducerEvent interface {
	//发送消息接口
	SendMessage(data []byte, key string) error

	//关闭生产者
	Close()
}

func NewAsyncProducer

func NewAsyncProducer(param ProducerParam) (ProducerEvent, error)

type ProducerParam

type ProducerParam struct {
	Address     []string
	Topic       string
	IsAck       bool
	KafkaVer    string
	Partitioner uint32
}