kafka

Documentation

Index

• Variables
• func JsonSerialize(v interface{}) ([]byte, error)
• func PBSerialize(v proto.Message) ([]byte, error)
• type Consumer
  • func NewConsumer(addrs []string, topic string) (*Consumer, error)
  • func NewConsumerWithLog(addrs []string, topic string, logPath string) (*Consumer, error)
  • func (c *Consumer) StartConsume()
• type Producer
  • func NewProducer(addrs []string, topic string, clientID string) (*Producer, error)
  • func NewProducerWithLog(addrs []string, topic, clientID, logPath, userName, passWord string, ...) (*Producer, error)
  • func (p *Producer) SendJsonWithKey(key string, msg interface{})
  • func (p *Producer) SendPBWithKey(key string, msg proto.Message)

Variables

var (
	MQMsgPool = &sync.Pool{
		New: func() interface{} {
			return &pb.MQData{}
		},
	}
)

Functions

func JsonSerialize

func JsonSerialize(v interface{}) ([]byte, error)

JsonSerialize

func PBSerialize

func PBSerialize(v proto.Message) ([]byte, error)

PBSerialize

Types

type Consumer

type Consumer struct {
	Consumer sarama.Consumer
	Topic    string
}

Consumer 消费者

func NewConsumer

func NewConsumer(addrs []string, topic string) (*Consumer, error)

func NewConsumerWithLog

func NewConsumerWithLog(addrs []string, topic string, logPath string) (*Consumer, error)

NewProducer creates kafka async producer

func (*Consumer) StartConsume

func (c *Consumer) StartConsume()

type Producer

type Producer struct {
	Producer sarama.AsyncProducer
	Topic    string
}

Producer represents the kafka async producer

func NewProducer

func NewProducer(addrs []string, topic string, clientID string) (*Producer, error)

NewProducer creates kafka async producer

func NewProducerWithLog

func NewProducerWithLog(addrs []string, topic, clientID, logPath, userName, passWord string, enableAuth bool) (*Producer, error)

NewProducer creates kafka async producer

func (*Producer) SendJsonWithKey

func (p *Producer) SendJsonWithKey(key string, msg interface{})

Send 发送

func (*Producer) SendPBWithKey

func (p *Producer) SendPBWithKey(key string, msg proto.Message)

Send 发送