kafka

type AsyncProducer interface {
	RunAsyncProducer()                                        // 运行异步生产者线程
	Produce(topic string, value []byte, keys ...string) error // 生产消息
	ProducerErrors() <-chan *sarama.ProducerError             // 返回生产者发送消息失败的chan
	CloseProducer()                                           // 关闭线程
	IsRunning() bool                                          // 运行状态
}

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

type Cli interface {
	Address() string                       // Address 返回kafka地址
	NewConsumer() (*ConsumerClient, error) // NewConsumer 新建消费者，kafka连接失败将返回error
	NewAsyncProducerClient() (AsyncProducer, error)
	NewSyncProducerClient() (sarama.SyncProducer, error)
}

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

type Config struct {
	Addr         string `yaml:"addr" env:"KafkaAddr" env-description:"address of kafka cluster"`
	QueueLength  int    `yaml:"queue_length"`
	KafkaVersion string `yaml:"kafka_version" env:"KafkaVersion" env-description:"version of kafka cluster"`
	EnableLog    bool   `yaml:"enable_log" env:"KafkaEnableLog" env-description:"enable kafka log or not"`
	LogLevel     string `yaml:"log_level" env:"KafkaLogLevel" env-description:"record logs in which level, only support debug/info"`
}

type Consumer interface {
	RunConsumer(group string, topic []string, handler sarama.ConsumerGroupHandler) error // 运行消费者线程
	Close()                                                                              // 关闭线程
	IsRunning() bool                                                                     // 运行状态
}

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

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