kafka

Documentation

Index

• Constants
• func Close()
• func SetupConsumer(opt *ConsumerConfig) (err error)
• func SetupProducer(opt *ProducerConfig) (err error)
• type ByteEncoder
• type Consumer
  • func GetConsumer(key string) Consumer
• type ConsumerConfig
• type ConsumerError
• type ConsumerErrorHandler
• type ConsumerMessage
• type ConsumerMessageHandler
• type Producer
  • func GetProducer(key string) Producer
• type ProducerConfig
• type ProducerError
• type ProducerErrorHandler
• type ProducerMessage
• type ProducerMessageHandler
• type StringEncoder

Constants

const (
	ACK_BEFORE_AUTO    = 0
	ACK_AFTER_NOERROR  = 1
	ACK_AFTER_NOMATTER = 2
)

Functions

func Close

func Close()

func SetupConsumer

func SetupConsumer(opt *ConsumerConfig) (err error)

func SetupProducer

func SetupProducer(opt *ProducerConfig) (err error)

Types

type ByteEncoder

type ByteEncoder = sarama.ByteEncoder

type Consumer

type Consumer interface {
	Close() error
	// blocking to consume the messages
	Consume(topics string, mh ConsumerMessageHandler, eh ConsumerErrorHandler) error
	ConsumeM(topics []string, mh ConsumerMessageHandler, eh ConsumerErrorHandler) error
}

func GetConsumer

func GetConsumer(key string) Consumer

type ConsumerConfig

type ConsumerConfig struct {
	Key          string
	Address      []string // kafka地址
	Group        string   // groupId
	Offset       int64
	Ack          int    // ack类型
	User         string //username and password for SASL/PLAIN  or SASL/SCRAM authentication
	Password     string
	DialTimeout  time.Duration // How long to wait for the initial connection.
	ReadTimeout  time.Duration // How long to wait for a response.
	WriteTimeout time.Duration // How long to wait for a transmit.
	KeepAlive    time.Duration
	Version      *sarama.KafkaVersion // kafka version
}

type ConsumerError

type ConsumerError = sarama.ConsumerError

type ConsumerErrorHandler

type ConsumerErrorHandler func(err error)

type ConsumerMessage

type ConsumerMessage = sarama.ConsumerMessage

type ConsumerMessageHandler

type ConsumerMessageHandler func(msg *ConsumerMessage) error

type Producer

type Producer interface {
	Close() error
	Produce(msgs ...*ProducerMessage) error
	AsyncHandle(mh ProducerMessageHandler, eh ProducerErrorHandler) // 必须设置 asyncReturnSuccess 或 asyncReturnError
}

func GetProducer

func GetProducer(key string) Producer

type ProducerConfig

type ProducerConfig struct {
	Key           string
	Address       []string // kafka地址
	Async         bool
	ReturnSuccess bool
	ReturnError   bool
	User          string //username and password for SASL/PLAIN  or SASL/SCRAM authentication
	Password      string
	Version       *sarama.KafkaVersion // kafka version
}

type ProducerError

type ProducerError = sarama.ProducerError

type ProducerErrorHandler

type ProducerErrorHandler func(err *ProducerError)

type ProducerMessage

type ProducerMessage = sarama.ProducerMessage

type ProducerMessageHandler

type ProducerMessageHandler func(msg *ProducerMessage)

type StringEncoder

type StringEncoder = sarama.StringEncoder