kfktool

README

kfktool

kfktool

Documentation

Index

• func PInit(path string) error
• func PInitWithConf(c *MyProducter) error
• type ConfProducter
• type Logger
  • func NewLogger(path, alias string, debug bool, args ...string) *Logger
  • func (l *Logger) CErr(format string, v ...interface{})
  • func (l *Logger) CInfo(format string, v ...interface{})
  • func (l *Logger) PErr(format string, v ...interface{})
  • func (l *Logger) PInfo(format string, v ...interface{})
• type MyConsumer
  • func NewConsumerGroup(brokers []string, topics []string, count int, groupName string) *MyConsumer
  • func (c *MyConsumer) Consumer(fn func(msg *sarama.ConsumerMessage) error)
  • func (c *MyConsumer) SetLogger(logger *Logger)
• type MyProducter
  • func (p *MyProducter) Close()
  • func (p *MyProducter) Dial() error
  • func (p *MyProducter) Publish(topic, value string, key ...string) (partition int32, offset int64, err error)
  • func (p *MyProducter) SetLogger(logger *Logger)

Functions

func PInit

func PInit(path string) error

PInit ... @path: 配置文件路径

func PInitWithConf

func PInitWithConf(c *MyProducter) error

PInitWithConf 通过配置文件初始化

Types

type ConfProducter

type ConfProducter []MyProducter

ConfProducter ...

type Logger

type Logger struct {
	LogConsumerSucc bool
	// contains filtered or unexported fields
}

Logger ...

func NewLogger

func NewLogger(path, alias string, debug bool, args ...string) *Logger

NewLogger ...

func (*Logger) CErr

func (l *Logger) CErr(format string, v ...interface{})

CErr consumer error

func (*Logger) CInfo

func (l *Logger) CInfo(format string, v ...interface{})

CInfo consumer info

func (*Logger) PErr

func (l *Logger) PErr(format string, v ...interface{})

PErr publish error

func (*Logger) PInfo

func (l *Logger) PInfo(format string, v ...interface{})

PInfo publish info

type MyConsumer

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

MyConsumer ...

func NewConsumerGroup

func NewConsumerGroup(brokers []string, topics []string, count int, groupName string) *MyConsumer

NewConsumerGroup 创建一个新的消费者组

func (*MyConsumer) Consumer

func (c *MyConsumer) Consumer(fn func(msg *sarama.ConsumerMessage) error)

Consumer 开始消费

func (*MyConsumer) SetLogger

func (c *MyConsumer) SetLogger(logger *Logger)

SetLogger SetLogger

type MyProducter

type MyProducter struct {
	Alias         string              `yaml:"alias"`
	Brokers       []string            `yaml:"brokers"`
	Sync          bool                `yaml:"sync"`
	WaitAck       sarama.RequiredAcks `yaml:"wait_ack"`
	SuccLog       bool                `yaml:"succ_log"`
	Debug         bool                `yaml:"debug"`
	SyncProducter sarama.SyncProducer
	AsyncProducer sarama.AsyncProducer
	// contains filtered or unexported fields
}

MyProducter ...

var (
	// P default producter
	P *MyProducter
	// PS producter map
	PS map[string]*MyProducter
)

func (*MyProducter) Close

func (p *MyProducter) Close()

Close ...

func (*MyProducter) Dial

func (p *MyProducter) Dial() error

Dial ...

func (*MyProducter) Publish

func (p *MyProducter) Publish(topic, value string, key ...string) (partition int32, offset int64, err error)

Publish ... 发消息

func (*MyProducter) SetLogger

func (p *MyProducter) SetLogger(logger *Logger)

SetLogger ... 使用自定的日志器