zdpgo_kafka

README

zdpgo_kafka

Golang操作kafka的组件

Documentation

Index

• type ConsumerGroupHandler
  • func (ConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error
  • func (c ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (ConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error
• type ConsumerGroupHandlerInterface
• type HandleGroupMessage
• type Kafka
  • func New(config KafkaConfig) *Kafka
  • func (k *Kafka) AsyncProducer(topic string, msg <-chan string)
  • func (k *Kafka) Close()
  • func (k *Kafka) ConsumePartition(topic string, partition int) (sarama.PartitionConsumer, error)
  • func (k *Kafka) GroupConsumer(groupName string, topics []string, handler ConsumerGroupHandlerInterface)
  • func (k *Kafka) IsDebug() bool
  • func (k *Kafka) Partitions(topic string) ([]int32, error)
  • func (k *Kafka) SendMessage(topic, value string) (partition int32, offset int64)
  • func (k *Kafka) SetDebug(debug bool)
• type KafkaConfig

Types

type ConsumerGroupHandler

type ConsumerGroupHandler struct {
	Log *zdpgo_log.Log // 日志记录对象
}

消费组处理器

func (ConsumerGroupHandler) Cleanup

func (ConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error

func (ConsumerGroupHandler) ConsumeClaim

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

ConsumeClaim 消费者组的处理器

func (ConsumerGroupHandler) Setup

func (ConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error

type ConsumerGroupHandlerInterface

type ConsumerGroupHandlerInterface interface {
	ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
	Setup(sarama.ConsumerGroupSession) error
	Cleanup(sarama.ConsumerGroupSession) error
}

消息处理接口

type HandleGroupMessage

type HandleGroupMessage func(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim)

处理消息

type Kafka

type Kafka struct {
	Consumer *sarama.Consumer     // 消费者
	Producer *sarama.SyncProducer // 生成者
	// contains filtered or unexported fields
}

核心类

func New

func New(config KafkaConfig) *Kafka

创建kafka对象的方法

func (*Kafka) AsyncProducer

func (k *Kafka) AsyncProducer(topic string, msg <-chan string)

异步生产者Goroutines

func (*Kafka) Close

func (k *Kafka) Close()

关闭kafka

func (*Kafka) ConsumePartition

func (k *Kafka) ConsumePartition(topic string, partition int) (sarama.PartitionConsumer, error)

根据分区创建消费者

func (*Kafka) GroupConsumer

func (k *Kafka) GroupConsumer(groupName string, topics []string, handler ConsumerGroupHandlerInterface)

消费者组

func (*Kafka) IsDebug

func (k *Kafka) IsDebug() bool

是否为debug模式

func (*Kafka) Partitions

func (k *Kafka) Partitions(topic string) ([]int32, error)

获取所有分区列表

func (*Kafka) SendMessage

func (k *Kafka) SendMessage(topic, value string) (partition int32, offset int64)

生产者发送消息

func (*Kafka) SetDebug

func (k *Kafka) SetDebug(debug bool)

设置debug模式

type KafkaConfig

type KafkaConfig struct {
	Host        string // 主机地址
	Port        int    // 端口号
	LogFilePath string // 日志文件路径
	Debug       bool   // 是否为调试模式
}

kafka配置信息