goo_kafka

Documentation

Index

• func Client() *client
• func Init(conf Config) error
• type Config
• type ConsumerError
• type ConsumerHandler
• type ConsumerMessage
• type IConsumer
  • func Consumer() IConsumer
• type IProducer
  • func Producer() IProducer

Functions

func Client

func Client() *client

func Init

func Init(conf Config) error

Types

type Config

type Config struct {
	User     string   `json:"user" yaml:"user"`
	Password string   `json:"password" yaml:"password"`
	Addrs    []string `json:"addrs" yaml:"addrs"`
	Timeout  int      `json:"timeout" yaml:"timeout"`
}

type ConsumerError

type ConsumerError struct {
	*sarama.ConsumerError
}

type ConsumerHandler

type ConsumerHandler func(msg *ConsumerMessage, err *ConsumerError) error

type ConsumerMessage

type ConsumerMessage struct {
	*sarama.ConsumerMessage
}

type IConsumer

type IConsumer interface {
	Close()

	// 处理分区消息
	// partition: 分区ID
	// offset: 消息偏移量，-2=从头开始，-1=获取最新的
	PartitionConsume(topic string, partition int32, offset int64, handler ConsumerHandler)

	// 处理指定偏移量消息
	// partition: 分区ID，默认为0
	// offset: 消息偏移量，-2=从头开始，-1=获取最新的
	Consume(topic string, offset int64, handler ConsumerHandler)

	// 处理最新消息
	// partition: 分区ID，默认为0
	// offset: 消息偏移量，-2=从头开始，-1=获取最新的
	ConsumeNewest(topic string, handler ConsumerHandler)

	// 处理所有消息，从第一条开始
	// partition: 分区ID，默认为0
	// offset: 消息偏移量，-2=从头开始，-1=获取最新的
	ConsumeOldest(topic string, handler ConsumerHandler)

	// 分组topic
	ConsumeGroup(groupId string, topics []string, handler ConsumerHandler)
	// contains filtered or unexported methods
}

消费者

func Consumer

func Consumer() IConsumer

type IProducer

type IProducer interface {
	Close()

	// 发送消息 - 同步
	SendMessage(key, topic string, message []byte) (partition int32, offset int64, err error)

	// 发送消息 - 异步
	SendAsyncMessage(key, topic string, message []byte) (partition int32, offset int64, err error)
	// contains filtered or unexported methods
}

生产者

func Producer

func Producer() IProducer