kafka

Documentation

Index

• type AsyncProducer
  • func (p *AsyncProducer) Produce(topic, value string)
• type Consumer
  • func (c *Consumer) Close()
  • func (c *Consumer) Consume(cb func(topic string, data []byte) bool) error
• type Kafka
  • func Ins(id ...string) *Kafka
  • func NewKafka(connect []string, config *cluster.Config) *Kafka
  • func (q *Kafka) AsyncProducer() (*AsyncProducer, error)
  • func (q *Kafka) Close() error
  • func (q *Kafka) NewConsumer(group string, topics ...string) (*Consumer, error)
  • func (q *Kafka) SyncProducer() (*SyncProducer, error)
• type SyncProducer
  • func (p *SyncProducer) Produce(topic, value string) (partition int32, offset int64, err error)

Types

type AsyncProducer

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

func (*AsyncProducer) Produce

func (p *AsyncProducer) Produce(topic, value string)

type Consumer

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

func (*Consumer) Close

func (c *Consumer) Close()

func (*Consumer) Consume

func (c *Consumer) Consume(cb func(topic string, data []byte) bool) error

type Kafka

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

func Ins

func Ins(id ...string) *Kafka

返回 kafka 组件单例

func NewKafka

func NewKafka(connect []string, config *cluster.Config) *Kafka

实例化一个全新的 Kafka

func (*Kafka) AsyncProducer

func (q *Kafka) AsyncProducer() (*AsyncProducer, error)

func (*Kafka) Close

func (q *Kafka) Close() error

func (*Kafka) NewConsumer

func (q *Kafka) NewConsumer(group string, topics ...string) (*Consumer, error)

func (*Kafka) SyncProducer

func (q *Kafka) SyncProducer() (*SyncProducer, error)

type SyncProducer

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

func (*SyncProducer) Produce

func (p *SyncProducer) Produce(topic, value string) (partition int32, offset int64, err error)