gokafka

README

kafka

• https://github.com/IBM/sarama

gokafka.Client().Topics()

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

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

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

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

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

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

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

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

Documentation

Index

• Constants
• func DelClient(names ...string)
• func Init(conf Config, opts ...Option) (err error)
• type Config
• type ConsumerError
• type ConsumerHandler
• type ConsumerMessage
  • func (msg ConsumerMessage) Commit()
• type GoKafka
  • func Client() *GoKafka
  • func GetClient(names ...string) *GoKafka
  • func New(conf Config, opts ...Option) (*GoKafka, error)
  • func (cli *GoKafka) Close()
  • func (cli *GoKafka) Consumer() IConsumer
  • func (cli *GoKafka) CreateTopicsRequest(topicName string, partitions int, replicationFactors int) error
  • func (c *GoKafka) GetKey(topic, msg string) string
  • func (c *GoKafka) OffsetInfo(topic, groupId string) (data []map[string]int64)
  • func (c *GoKafka) Partitions(topic string) []int32
  • func (cli *GoKafka) Producer(opts ...Option) IProducer
  • func (c *GoKafka) Redis() *goredis.GoRedis
  • func (c *GoKafka) Topics() []string
• type IConsumer
  • func Consumer() IConsumer
• type IMessage
• type IProducer
  • func Producer(opts ...Option) IProducer
• type MessageHandler
• type Option
  • func FocusOption() Option
  • func RedisOption(cli *goredis.GoRedis) Option
• type ProducerMessage

Constants

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

Functions

func DelClient

func DelClient(names ...string)

func Init

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

Types

type Config

type Config struct {
	User              string   `json:"user,optional" yaml:"User"`
	Name              string   `json:"name,optional"  yaml:"Name"`
	Password          string   `json:"password,optional"  yaml:"Password"`
	Addrs             []string `json:"addrs,optional"  yaml:"Addrs"`
	Timeout           int      `json:"timeout,optional"  yaml:"Timeout"`
	HeartbeatInterval int      `json:"heartbeatInterval,optional" yaml:"HeartbeatInterval"`
	SessionTimeout    int      `json:"sessionTimeout,optional" yaml:"SessionTimeout"`
	RebalanceTimeout  int      `json:"rebalanceTimeout,optional" yaml:"RebalanceTimeout"`
	OffsetNewest      bool     `json:"offsetNewest,optional" yaml:"OffsetNewest"`
	GroupId           string   `json:"groupId,optional" yaml:"GroupId"`

	Version string `json:"version,optional" yaml:"Version"`

	RedisConfig goredis.Config `json:"redisConfig,optional" yaml:"RedisConfig"`
}

type ConsumerError

type ConsumerError struct {
	*sarama.ConsumerError
}

type ConsumerHandler

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

type ConsumerMessage

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

func (ConsumerMessage) Commit

func (msg ConsumerMessage) Commit()

type GoKafka

type GoKafka struct {
	sarama.Client
	// contains filtered or unexported fields
}

func Client

func Client() *GoKafka

default or 只有一个kafka实例直接返回

func GetClient

func GetClient(names ...string) *GoKafka

func New

func New(conf Config, opts ...Option) (*GoKafka, error)

func (*GoKafka) Close

func (cli *GoKafka) Close()

func (*GoKafka) Consumer

func (cli *GoKafka) Consumer() IConsumer

消费者

func (*GoKafka) CreateTopicsRequest

func (cli *GoKafka) CreateTopicsRequest(topicName string, partitions int, replicationFactors int) error

func (*GoKafka) GetKey

func (c *GoKafka) GetKey(topic, msg string) string

func (*GoKafka) OffsetInfo

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

分区数量

func (*GoKafka) Partitions

func (c *GoKafka) Partitions(topic string) []int32

分区数量

func (*GoKafka) Producer

func (cli *GoKafka) Producer(opts ...Option) IProducer

生产者

func (*GoKafka) Redis

func (c *GoKafka) Redis() *goredis.GoRedis

func (*GoKafka) Topics

func (c *GoKafka) Topics() []string

主题列表

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 *goredis.GoRedis) Option

redis 对象

type ProducerMessage

type ProducerMessage struct {
	*sarama.ProducerMessage
}