mkafka

func CreateKafkaClient ¶

func CreateKafkaClient(brokers string, config *sarama.Config) (sarama.Client, error)

CreateKafkaClient 创建kafka客户端，使用后需关闭

func CreateTopic ¶

func CreateTopic(client sarama.Client, topic string, partition int32, replica int16) error

CreateTopic 创建topic

func CustomConsumerGroup ¶

func CustomConsumerGroup(brokers string, group string, config *sarama.Config) (sarama.ConsumerGroup, error)

CustomConsumerGroup 根据传入的config参数创建消费者组

func DefaultConsumerConfig ¶

func DefaultConsumerConfig() *sarama.Config

DefaultConsumerConfig 默认消费者组配置

func DefaultConsumerGroup ¶

func DefaultConsumerGroup(brokers string, group string) (sarama.ConsumerGroup, error)

DefaultConsumerGroup 默认消费者组

func DefaultProducer ¶

func DefaultProducer(brokers string) (sarama.SyncProducer, error)

DefaultProducer 默认生产者

func DefaultProducerConfig ¶

func DefaultProducerConfig() *sarama.Config

DefaultProducerConfig 默认生产者配置

func DeleteConsumerGroup ¶

func DeleteConsumerGroup(client sarama.Client, groups []string) error

func RemoveTopic ¶

func RemoveTopic(client sarama.Client, topics []string) error

RemoveTopic 删除topic

func ResetConsumerGroupOffset ¶

func ResetConsumerGroupOffset(client sarama.Client, group string, allTopic bool, partitions map[string][]int32) error

ResetConsumerGroupOffset 通过消费者组删除对某个topic的某些partition的offset

func SetLogger ¶

func SetLogger(out zerolog.Logger, saraLog bool)