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 Client() *client
• func Consumer() iConsumer
• func GetClient(names ...string) *client
• func Init(configs ...Config) (err error)
• func New(conf Config) (*client, error)
• func Producer() iProducer
• type Config
• type ConsumerError
• type ConsumerHandler
• type ConsumerMessage
• type MessageHandler
• type ProducerMessage

Functions

func Client

func Client() *client

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

func Consumer

func Consumer() iConsumer

func GetClient

func GetClient(names ...string) *client

func Init

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

func New

func New(conf Config) (*client, 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 MessageHandler

type MessageHandler func(msg *ProducerMessage, err error)

type ProducerMessage

type ProducerMessage struct {
	*sarama.ProducerMessage
}