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

• func Consumer() iConsumer
• func DelClient(names ...string)
• func Init(configs ...Config) (err error)
• func Producer() iProducer
• type Config
• type ConsumerError
• type ConsumerHandler
• type ConsumerMessage
• type GoKafka
  • func Client() *GoKafka
  • func GetClient(names ...string) *GoKafka
  • func New(conf Config) (*GoKafka, error)
  • func (cli *GoKafka) Close()
  • func (cli *GoKafka) Consumer() iConsumer
  • func (cli *GoKafka) CreateTopicsRequest(topicName string, partitions int, replicationFactors int) error
  • func (cli *GoKafka) Producer() iProducer
• type MessageHandler
• type ProducerMessage

Functions

func Consumer

func Consumer() iConsumer

func DelClient

func DelClient(names ...string)

func Init

func Init(configs ...Config) (err error)

func Producer

func Producer() iProducer

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" yaml:"HeartbeatInterval"`
	SessionTimeout    int      `json:"sessionTimeout" yaml:"SessionTimeout"`
	RebalanceTimeout  int      `json:"rebalanceTimeout" yaml:"RebalanceTimeout"`
	OffsetNewest      bool     `json:"offsetNewest" yaml:"OffsetNewest"`
}

type ConsumerError

type ConsumerError struct {
	*sarama.ConsumerError
}

type ConsumerHandler

type ConsumerHandler func(msg *ConsumerMessage, consumerErr *ConsumerError) error

type ConsumerMessage

type ConsumerMessage struct {
	*sarama.ConsumerMessage
}

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) (*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) Producer

func (cli *GoKafka) Producer() iProducer

type MessageHandler

type MessageHandler func(msg *ProducerMessage, err error)

type ProducerMessage

type ProducerMessage struct {
	*sarama.ProducerMessage
}