kafkaHelper

Documentation

Index

• func AsyncProductMsg(topic string, msgs []string)
• func Send2Topic(brokers []string, topic, content string)
• type ConsumerGroupHandler
  • func NewConsumerGroupHandler(Brokers []string, GroupId string) *ConsumerGroupHandler
  • func (h *ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error
  • func (h *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, Claim sarama.ConsumerGroupClaim) error
  • func (h *ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error
  • func (h *ConsumerGroupHandler) Subscribe(topic string, fun func(string) bool)
  • func (h *ConsumerGroupHandler) SubscribeRaw(topic string, fun func(*sarama.ConsumerMessage) bool)
  • func (h *ConsumerGroupHandler) Worker()
• type KafkaAsyncProducer
  • func NewKafkaAsyncProducer(brokers []string) *KafkaAsyncProducer
  • func (p *KafkaAsyncProducer) Close()
  • func (p *KafkaAsyncProducer) Publish(topic string, messageBytes []byte)

Functions

func AsyncProductMsg

func AsyncProductMsg(topic string, msgs []string)

AsyncProductMsg 是一个异步发送消息到Kafka的函数 topic: 消息的主题 msgs: 需要发送的消息的内容列表

func Send2Topic

func Send2Topic(brokers []string, topic, content string)

同步生产模式

Types

type ConsumerGroupHandler

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

func NewConsumerGroupHandler

func NewConsumerGroupHandler(Brokers []string, GroupId string) *ConsumerGroupHandler

func (*ConsumerGroupHandler) Cleanup

func (h *ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error

func (*ConsumerGroupHandler) ConsumeClaim

func (h *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, Claim sarama.ConsumerGroupClaim) error

func (*ConsumerGroupHandler) Setup

func (h *ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error

func (*ConsumerGroupHandler) Subscribe

func (h *ConsumerGroupHandler) Subscribe(topic string, fun func(string) bool)

func (*ConsumerGroupHandler) SubscribeRaw

func (h *ConsumerGroupHandler) SubscribeRaw(topic string, fun func(*sarama.ConsumerMessage) bool)

func (*ConsumerGroupHandler) Worker

func (h *ConsumerGroupHandler) Worker()

type KafkaAsyncProducer

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

func NewKafkaAsyncProducer

func NewKafkaAsyncProducer(brokers []string) *KafkaAsyncProducer

func (*KafkaAsyncProducer) Close

func (p *KafkaAsyncProducer) Close()

func (*KafkaAsyncProducer) Publish

func (p *KafkaAsyncProducer) Publish(topic string, messageBytes []byte)