kafka2

Documentation

Index

• Constants
• type Consumer
  • func NewConsumer() *Consumer
  • func (c *Consumer) Close() error
  • func (c *Consumer) Init(v interface{}) error
  • func (c *Consumer) LoadConfig(f string) error
  • func (c *Consumer) Read(p []byte) (cnt int, err error)
  • func (c *Consumer) Start(ctx context.Context) error
  • func (c *Consumer) Version() string

Constants

const (
	// VERSION version
	VERSION = "0.0.1"
)

Types

type Consumer

type Consumer struct {
	Addrs           []string    `json:"addrs" yaml:"addrs"` //如果定义了group,则addrs是zookeeper的地址(2181)，否则的话是kafka的地址(9092)
	Topics          []string    `json:"topics" yaml:"topics"`
	Group           string      `json:"group" yaml:"group"`
	Offset          string      `json:"offset" yaml:"offset"`
	Message         chan []byte `json:"-" yaml:"-"` //从这个管道中读取数据
	*kafka.Consumer `json:"-" yaml:"-"`
}

Consumer xx 一些使用说明: sarame.OffsetNewest int64 = -1 sarame.OffsetOldest int64 = -2

func NewConsumer

func NewConsumer() *Consumer

NewConsumer kafka input

func (*Consumer) Close

func (c *Consumer) Close() error

Close close

func (*Consumer) Init

func (c *Consumer) Init(v interface{}) error

Init init

func (*Consumer) LoadConfig

func (c *Consumer) LoadConfig(f string) error

LoadConfig load config

func (*Consumer) Read

func (c *Consumer) Read(p []byte) (cnt int, err error)

func (*Consumer) Start

func (c *Consumer) Start(ctx context.Context) error

Start start

func (*Consumer) Version

func (c *Consumer) Version() string

Version version