kafka

Documentation

Index

• func MessageToProducerMessage(msg *Message) (producerMessage *sarama.ProducerMessage, err error)
• type BrokerInfo
• type Config
• type IService
  • func New(config Config) (IService, error)
• type Info
• type Message
  • func ConsumerMessageToMessage(keyType string, valueType string, consumerMessage *sarama.ConsumerMessage) (msg *Message, err error)
• type MessageHeader
• type ProducerMessage
• type Service
  • func (this_ *Service) CreatePartitions(topic string, count int32) (err error)
  • func (this_ *Service) CreateTopic(topic string, numPartitions int32, replicationFactor int16) (err error)
  • func (this_ *Service) DeleteConsumerGroup(groupId string) (err error)
  • func (this_ *Service) DeleteRecords(topic string, partitionOffsets map[int32]int64) (err error)
  • func (this_ *Service) DeleteTopic(topic string) (err error)
  • func (this_ *Service) GetServers() []string
  • func (this_ *Service) GetTopics() (res []*TopicInfo, err error)
  • func (this_ *Service) Info() (res *Info, err error)
  • func (this_ *Service) MarkOffset(groupId string, topic string, partition int32, offset int64) (err error)
  • func (this_ *Service) NewSyncProducer() (syncProducer sarama.SyncProducer, err error)
  • func (this_ *Service) Pull(groupId string, topics []string, PullSize int, PullTimeout int, ...) (msgList []*Message, err error)
  • func (this_ *Service) Push(msg *Message) (err error)
  • func (this_ *Service) ResetOffset(groupId string, topic string, partition int32, offset int64) (err error)
  • func (this_ *Service) Stop()
• type TopicInfo

Functions

func MessageToProducerMessage

func MessageToProducerMessage(msg *Message) (producerMessage *sarama.ProducerMessage, err error)

Types

type BrokerInfo

type BrokerInfo struct {
	Id        int32  `json:"id"`
	Addr      string `json:"addr"`
	Rack      string `json:"rack"`
	Connected bool   `json:"connected"`
}

type Config

type Config struct {
	Address  string `json:"address"`
	Username string `json:"username,omitempty"`
	Password string `json:"password,omitempty"`
	CertPath string `json:"certPath,omitempty"`
}

Config kafka配置

type IService

type IService interface {
	Stop()
	Info() (res *Info, err error)
	GetTopics() (res []*TopicInfo, err error)
	Pull(groupId string, topics []string, PullSize int, PullTimeout int, keyType, valueType string) (msgList []*Message, err error)
	MarkOffset(groupId string, topic string, partition int32, offset int64) (err error)
	ResetOffset(groupId string, topic string, partition int32, offset int64) (err error)
	CreatePartitions(topic string, count int32) (err error)
	CreateTopic(topic string, numPartitions int32, replicationFactor int16) (err error)
	DeleteTopic(topic string) (err error)
	DeleteConsumerGroup(groupId string) (err error)
	DeleteRecords(topic string, partitionOffsets map[int32]int64) (err error)
	NewSyncProducer() (syncProducer sarama.SyncProducer, err error)
	Push(msg *Message) (err error)
}

func New

func New(config Config) (IService, error)

New 创建kafka服务

type Info

type Info struct {
	Brokers []*BrokerInfo `json:"brokers"`
}

type Message

type Message struct {
	KeyType   string          `json:"keyType,omitempty"`
	Key       string          `json:"key,omitempty"`
	ValueType string          `json:"valueType,omitempty"`
	Value     string          `json:"value,omitempty"`
	Topic     string          `json:"topic,omitempty"`
	Partition *int32          `json:"partition,omitempty"`
	Offset    *int64          `json:"offset,omitempty"`
	Headers   []MessageHeader `json:"headers,omitempty"`
	Timestamp *time.Time      `json:"timestamp,omitempty"`
}

func ConsumerMessageToMessage

func ConsumerMessageToMessage(keyType string, valueType string, consumerMessage *sarama.ConsumerMessage) (msg *Message, err error)

type MessageHeader

type MessageHeader struct {
	Key   string `json:"key,omitempty"`
	Value string `json:"value,omitempty"`
}

type ProducerMessage

type ProducerMessage struct {
	*sarama.ProducerMessage
}

type Service

type Service struct {
	Config
}

Service 注册处理器在线信息等

func (*Service) CreatePartitions

func (this_ *Service) CreatePartitions(topic string, count int32) (err error)

func (*Service) CreateTopic

func (this_ *Service) CreateTopic(topic string, numPartitions int32, replicationFactor int16) (err error)

func (*Service) DeleteConsumerGroup

func (this_ *Service) DeleteConsumerGroup(groupId string) (err error)

func (*Service) DeleteRecords

func (this_ *Service) DeleteRecords(topic string, partitionOffsets map[int32]int64) (err error)

func (*Service) DeleteTopic

func (this_ *Service) DeleteTopic(topic string) (err error)

func (*Service) GetServers

func (this_ *Service) GetServers() []string

func (*Service) GetTopics

func (this_ *Service) GetTopics() (res []*TopicInfo, err error)

func (*Service) Info

func (this_ *Service) Info() (res *Info, err error)

func (*Service) MarkOffset

func (this_ *Service) MarkOffset(groupId string, topic string, partition int32, offset int64) (err error)

func (*Service) NewSyncProducer

func (this_ *Service) NewSyncProducer() (syncProducer sarama.SyncProducer, err error)

NewSyncProducer 创建生产者

func (*Service) Pull

func (this_ *Service) Pull(groupId string, topics []string, PullSize int, PullTimeout int, keyType, valueType string) (msgList []*Message, err error)

func (*Service) Push

func (this_ *Service) Push(msg *Message) (err error)

Push 推送消息到kafka

func (*Service) ResetOffset

func (this_ *Service) ResetOffset(groupId string, topic string, partition int32, offset int64) (err error)

func (*Service) Stop

func (this_ *Service) Stop()

type TopicInfo

type TopicInfo struct {
	Topic string `json:"topic"`
}