kafka

Documentation

Index

• Constants
• func OperationIDGenerator() string
• type BatchAggregationIdListHandlerF
• type BatchConsumerGroup
  • func MustKafkaBatchConsumer(c *KafkaConsumerConf) *BatchConsumerGroup
  • func (c *BatchConsumerGroup) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *BatchConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *BatchConsumerGroup) Group() string
  • func (c *BatchConsumerGroup) MessagesDistributionHandle()
  • func (c *BatchConsumerGroup) RegisterHandlers(cb BatchAggregationIdListHandlerF, cb2 BatchMessageHandlerF)
  • func (c *BatchConsumerGroup) Run(channelID int)
  • func (c *BatchConsumerGroup) Setup(sarama.ConsumerGroupSession) error
  • func (c *BatchConsumerGroup) Start()
  • func (c *BatchConsumerGroup) Stop()
  • func (c *BatchConsumerGroup) Topics() []string
• type BatchMessageHandlerF
• type Cmd2Value
• type ConsumerGroup
  • func MustKafkaConsumer(c *KafkaConsumerConf) *ConsumerGroup
  • func (c *ConsumerGroup) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *ConsumerGroup) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *ConsumerGroup) Group() string
  • func (c *ConsumerGroup) RegisterHandlers(topic string, cb MessageHandlerF)
  • func (c *ConsumerGroup) Setup(sarama.ConsumerGroupSession) error
  • func (c *ConsumerGroup) Start()
  • func (c *ConsumerGroup) Stop()
  • func (c *ConsumerGroup) Topics() []string
• type KafkaBatchConsumerConf
• type KafkaConsumerConf
• type KafkaProducerConf
• type MessageHandlerF
• type MsgChannelValue
• type MsgDataToMQ
• type Producer
  • func GetCachedMQClient(c *KafkaProducerConf) *Producer
  • func MustKafkaProducer(c *KafkaProducerConf) *Producer
  • func (p *Producer) Close() (err error)
  • func (p *Producer) SendMessage(ctx context.Context, key string, value []byte) (partition int32, offset int64, err error)
  • func (p *Producer) Topic() string
• type TriggerChannelValue

Constants

const (
	ConsumerMsgs        = 3
	AggregationMessages = 4
	ChannelNum          = 100
)

Functions

func OperationIDGenerator

func OperationIDGenerator() string

Types

type BatchAggregationIdListHandlerF

type BatchAggregationIdListHandlerF func(triggerID string, idList []string)

type BatchConsumerGroup

type BatchConsumerGroup struct {
	sarama.ConsumerGroup
	// contains filtered or unexported fields
}

BatchConsumerGroup kafka consumer

func MustKafkaBatchConsumer

func MustKafkaBatchConsumer(c *KafkaConsumerConf) *BatchConsumerGroup

func (*BatchConsumerGroup) Cleanup

func (c *BatchConsumerGroup) Cleanup(sarama.ConsumerGroupSession) error

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

func (*BatchConsumerGroup) ConsumeClaim

func (c *BatchConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*BatchConsumerGroup) Group

func (c *BatchConsumerGroup) Group() string

func (*BatchConsumerGroup) MessagesDistributionHandle

func (c *BatchConsumerGroup) MessagesDistributionHandle()

func (*BatchConsumerGroup) RegisterHandlers

func (c *BatchConsumerGroup) RegisterHandlers(cb BatchAggregationIdListHandlerF, cb2 BatchMessageHandlerF)

func (*BatchConsumerGroup) Run

func (c *BatchConsumerGroup) Run(channelID int)

func (*BatchConsumerGroup) Setup

func (c *BatchConsumerGroup) Setup(sarama.ConsumerGroupSession) error

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

func (*BatchConsumerGroup) Start

func (c *BatchConsumerGroup) Start()

Start start consume messages, watch signals

func (*BatchConsumerGroup) Stop

func (c *BatchConsumerGroup) Stop()

Stop Stop consume messages, watch signals

func (*BatchConsumerGroup) Topics

func (c *BatchConsumerGroup) Topics() []string

type BatchMessageHandlerF

type BatchMessageHandlerF func(ctx context.Context, msgs MsgChannelValue)

type Cmd2Value

type Cmd2Value struct {
	Cmd   int
	Value interface{}
}

type ConsumerGroup

type ConsumerGroup struct {
	sarama.ConsumerGroup
	// contains filtered or unexported fields
}

ConsumerGroup kafka consumer

func MustKafkaConsumer

func MustKafkaConsumer(c *KafkaConsumerConf) *ConsumerGroup

func (*ConsumerGroup) Cleanup

func (c *ConsumerGroup) Cleanup(sarama.ConsumerGroupSession) error

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

func (*ConsumerGroup) ConsumeClaim

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

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*ConsumerGroup) Group

func (c *ConsumerGroup) Group() string

func (*ConsumerGroup) RegisterHandlers

func (c *ConsumerGroup) RegisterHandlers(topic string, cb MessageHandlerF)

func (*ConsumerGroup) Setup

func (c *ConsumerGroup) Setup(sarama.ConsumerGroupSession) error

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

func (*ConsumerGroup) Start

func (c *ConsumerGroup) Start()

Start start consume messages, watch signals

func (*ConsumerGroup) Stop

func (c *ConsumerGroup) Stop()

Stop Stop consume messages, watch signals

func (*ConsumerGroup) Topics

func (c *ConsumerGroup) Topics() []string

type KafkaBatchConsumerConf

type KafkaBatchConsumerConf struct {
	KafkaConsumerConf
	Duration   int `json:",default=100"`
	ChannelNum int `json:"default=50"`
}

KafkaBatchConsumerConf kafka client settings.

type KafkaConsumerConf

type KafkaConsumerConf struct {
	Brokers []string
	Topics  []string
	Group   string
}

KafkaConsumerConf kafka client settings.

type KafkaProducerConf

type KafkaProducerConf struct {
	Brokers []string
	Topic   string
}

KafkaProducerConf kafka producer settings.

type MessageHandlerF

type MessageHandlerF func(ctx context.Context, key string, value []byte)

type MsgChannelValue

type MsgChannelValue struct {
	AggregationID string //maybe userID or super groupID
	TriggerID     string
	MsgList       []*MsgDataToMQ
}

type MsgDataToMQ

type MsgDataToMQ struct {
	TraceId string
	MsgType string
	MsgData []byte
}

type Producer

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

func GetCachedMQClient

func GetCachedMQClient(c *KafkaProducerConf) *Producer

func MustKafkaProducer

func MustKafkaProducer(c *KafkaProducerConf) *Producer

func (*Producer) Close

func (p *Producer) Close() (err error)

func (*Producer) SendMessage

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

SendMessage Input send msg to kafka NOTE: If producer has beed created failed, the message will lose.

func (*Producer) Topic

func (p *Producer) Topic() string

type TriggerChannelValue

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