kafka

Documentation

Index

• func SetTrace(t bool)
• type ClientReader
  • func (cr *ClientReader) ReadMessage(ctx context.Context) (kafka.Message, error)
• type ClientWriter
  • func (cw *ClientWriter) SendMessage(ctx context.Context, msg ...kafka.Message) error
• type Clientkafka
  • func New(config *Config) *Clientkafka
  • func (client *Clientkafka) CreateTopic(topic string) error
  • func (client *Clientkafka) Host() string
  • func (client *Clientkafka) Kafka() *kafka.Conn
  • func (client *Clientkafka) ListTopics() []string
  • func (client *Clientkafka) NewReader(topic string) *ClientReader
  • func (client *Clientkafka) NewWriter(topic string) *ClientWriter
• type Config
• type Proxy
  • func NewProxy() *Proxy
  • func (proxy *Proxy) NameSpace(group string) *Clientkafka
  • func (proxy *Proxy) SetNameSpace(group string, client *Clientkafka)
• type Reader
  • func (proxy *Reader) NameSpace(topic string) *ClientReader
  • func (proxy *Reader) SetNameSpace(group string, client *ClientReader)
• type Writer
  • func (proxy *Writer) NameSpace(topic string) *ClientWriter
  • func (proxy *Writer) SetNameSpace(group string, client *ClientWriter)

Functions

func SetTrace

func SetTrace(t bool)

Types

type ClientReader

type ClientReader struct {
	*kafka.Reader
}

func (*ClientReader) ReadMessage

func (cr *ClientReader) ReadMessage(ctx context.Context) (kafka.Message, error)

type ClientWriter

type ClientWriter struct {
	*kafka.Writer
}

func (*ClientWriter) SendMessage

func (cw *ClientWriter) SendMessage(ctx context.Context, msg ...kafka.Message) error

type Clientkafka

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

func New

func New(config *Config) *Clientkafka

New 新连接

func (*Clientkafka) CreateTopic

func (client *Clientkafka) CreateTopic(topic string) error

CreateTopic create

func (*Clientkafka) Host

func (client *Clientkafka) Host() string

func (*Clientkafka) Kafka

func (client *Clientkafka) Kafka() *kafka.Conn

func (*Clientkafka) ListTopics

func (client *Clientkafka) ListTopics() []string

ListTopics To list topics

func (*Clientkafka) NewReader

func (client *Clientkafka) NewReader(topic string) *ClientReader

func (*Clientkafka) NewWriter

func (client *Clientkafka) NewWriter(topic string) *ClientWriter

NewWriter ..

type Config

type Config struct {
	Username string //账号 root
	Password string //密码
	Host     string //host localhost
	Broker   []string
}

Config 数据库配置

type Proxy

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

Proxy 代理

func NewProxy

func NewProxy() *Proxy

NewProxy 代理池

func (*Proxy) NameSpace

func (proxy *Proxy) NameSpace(group string) *Clientkafka

NameSpace 获取分组

func (*Proxy) SetNameSpace

func (proxy *Proxy) SetNameSpace(group string, client *Clientkafka)

SetNameSpace 设置组

type Reader

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

func (*Reader) NameSpace

func (proxy *Reader) NameSpace(topic string) *ClientReader

NameSpace 获取分组

func (*Reader) SetNameSpace

func (proxy *Reader) SetNameSpace(group string, client *ClientReader)

SetNameSpace 设置组

type Writer

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

func (*Writer) NameSpace

func (proxy *Writer) NameSpace(topic string) *ClientWriter

NameSpace 获取分组

func (*Writer) SetNameSpace

func (proxy *Writer) SetNameSpace(group string, client *ClientWriter)

SetNameSpace 设置组