Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
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 (*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 ¶
Source Files
Click to show internal directories.
Click to hide internal directories.