kafka

Documentation

Index

• type Client
  • func NewClient(topics []string, conf *Config, log sarama.StdLogger) (client *Client, e error)
  • func (p *Client) RecvMsg() ([]byte, error)
  • func (p *Client) SendMsg(topic string, msgIn []byte) error
  • func (p *Client) SendMsgAsync(topic string, msgIn []byte) error
• type Config

Types

type Client

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

func NewClient

func NewClient(topics []string, conf *Config, log sarama.StdLogger) (client *Client, e error)

NewClient 新建Kafka需要通过config，所有必须通过此函数创建； 有一个问题，logger无法在请求的时候返回，是否需要一个全局的logger？

func (*Client) RecvMsg

func (p *Client) RecvMsg() ([]byte, error)

func (*Client) SendMsg

func (p *Client) SendMsg(topic string, msgIn []byte) error

同步发送接口

func (*Client) SendMsgAsync

func (p *Client) SendMsgAsync(topic string, msgIn []byte) error

同步发送接口

type Config

type Config struct {
	ServerList  []string // kafka服务器地址
	SendTimeOut int      // 发送超时时间， 单位毫秒
	RetryTime   int      // 重传次数， 消息系统里面快速失败，不保证数据的高可到达性
	RetryWait   int      // 重传的等待时间, 单位毫秒
	Print       int      // 错误打印在串口
	OffSet      int64    // 启动时，接收消息的位置， -1 从最新的开始读， -2：从上次的问题开始读
	Group       string   // 读消息的组配置
}