goo_kafka

README

kafka

goo_kafka.Client().Topics()

// 发送消息，不指定分区
goo_kafka.Producer().SendMessage("test", []byte("hi hnatao"))

// 发送消息，指定分区
goo_kafka.Producer().WithPartition(0).SendMessage("test", []byte("hi hnatao"))

// 发送异步消息，不指定分区
goo_kafka.Producer().SendAsyncMessage("test", []byte("hi hnatao"), func(msg *goo_kafka.ProducerMessage, err error) {
})

// 发送异步消息，指定分区
goo_kafka.Producer().WithPartition(0).SendAsyncMessage("test", []byte("hi hnatao"), func(msg *goo_kafka.ProducerMessage, err error) {
})

// 消费消息，指定分区，指定起始位置
goo_kafka.Consumer().WithPartition(0).WithOffset(100).Consume("test", func(msg *goo_kafka.ConsumerMessage, consumerErr *goo_kafka.ConsumerError) error {
    return nil
})

// 消费消息，指定分区，从最新位置开始
goo_kafka.Consumer().WithPartition(0).WithOffsetNewest().Consume("test", func(msg *goo_kafka.ConsumerMessage, consumerErr *goo_kafka.ConsumerError) error {
    return nil
})

// 消费消息，指定分区，从最头开始
goo_kafka.Consumer().WithPartition(0).WithOffsetOldest().Consume("test", func(msg *goo_kafka.ConsumerMessage, consumerErr *goo_kafka.ConsumerError) error {
    return nil
})

// 消费消息，分组消息，分组里面只要1个消费者消费
goo_kafka.Consumer().ConsumeGroup("test-id", []string{"test"}, func(msg *goo_kafka.ConsumerMessage, consumerErr *goo_kafka.ConsumerError) error {
    return nil
})

Documentation

Index

• Constants
• func Client() *client
• func Init(conf Config, opts ...Option) error
• func OffsetInfo(topic, groupId string) (data []map[string]int64)
• func Partitions(topic string) []int32
• func Topics() []string
• type Config
• type ConsumerError
• type ConsumerHandler
• type ConsumerMessage
  • func (msg ConsumerMessage) Commit()
• type IConsumer
  • func Consumer() IConsumer
• type IMessage
• type IProducer
  • func Producer(opts ...Option) IProducer
• type MessageHandler
• type Option
  • func FocusOption() Option
  • func RedisOption(cli *goo_redis.Client) Option
• type ProducerMessage

Constants

const (
	FocusName = "focus"
	RedisName = "redis"
)

Functions

func Client

func Client() *client

客户端

func Init

func Init(conf Config, opts ...Option) error

初始化

func OffsetInfo

func OffsetInfo(topic, groupId string) (data []map[string]int64)

分区数量

func Partitions

func Partitions(topic string) []int32

分区数量

func Topics

func Topics() []string

主题列表

Types

type Config

type Config struct {
	User              string           `json:"user" yaml:"user"`
	Password          string           `json:"password" yaml:"password"`
	Addrs             []string         `json:"addrs" yaml:"addrs"`
	Timeout           int              `json:"timeout" yaml:"timeout"`
	HeartbeatInterval int              `json:"heartbeat_interval" yaml:"heartbeat_interval"`
	SessionTimeout    int              `json:"session_timeout" yaml:"session_timeout"`
	RebalanceTimeout  int              `json:"rebalance_timeout" yaml:"rebalance_timeout"`
	RedisConfig       goo_redis.Config `json:"redis_config" yaml:"redis_config"`
}

type ConsumerError

type ConsumerError struct {
	*sarama.ConsumerError
}

type ConsumerHandler

type ConsumerHandler func(ctx *goo_context.Context, msg *ConsumerMessage, consumerErr *ConsumerError) error

type ConsumerMessage

type ConsumerMessage struct {
	*sarama.ConsumerMessage
	GroupSession sarama.ConsumerGroupSession
}

func (ConsumerMessage) Commit

func (msg ConsumerMessage) Commit()

type IConsumer

type IConsumer interface {
	Client() sarama.Client

	// 从指定分区消费
	WithPartition(partition int32) IConsumer

	// 从指定位置开始
	WithOffset(offset int64) IConsumer

	// 从最新位置开始
	WithOffsetNewest() IConsumer

	// 从头开始
	WithOffsetOldest() IConsumer

	// 消费
	Consume(topic string, handler ConsumerHandler)

	// 分组topic
	ConsumeGroup(groupId string, topics []string, handler ConsumerHandler)
}

消费者

func Consumer

func Consumer() IConsumer

消费者

type IMessage

type IMessage interface {
	Topic() string
	Key() string
	Headers() map[string]string
	Serialize() []byte
	Deserialize(b []byte)
}

消息

type IProducer

type IProducer interface {
	Client() sarama.Client

	// 发送消息 - 同步
	SendMessage(msg IMessage) (partition int32, offset int64, err error)

	// 发送消息 - 异步
	SendAsyncMessage(msg IMessage, cb MessageHandler) (err error)
}

生产者

func Producer

func Producer(opts ...Option) IProducer

生产者

type MessageHandler

type MessageHandler func(msg *ProducerMessage, err error)

type Option

type Option struct {
	Name  string
	Value interface{}
}

func FocusOption

func FocusOption() Option

是否强制

func RedisOption

func RedisOption(cli *goo_redis.Client) Option

redis

type ProducerMessage

type ProducerMessage struct {
	*sarama.ProducerMessage
}