kafka

README

更简单易用的kafka客户端

参考 B站代码

• https://github.com/go-kratos/kafka

Documentation

Index

• Constants
• Variables
• type Config
• type Event
• type Handler
• type Publisher
  • func NewPublisher(config *Config, opts ...PublisherOption) *Publisher
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(ctx context.Context, event Event) error
• type PublisherOption
  • func BatchSize(size int) PublisherOption
  • func EventBuffer(n int) PublisherOption
  • func ReadTimeout(d time.Duration) PublisherOption
  • func RequiredAcks(acks kafka.RequiredAcks) PublisherOption
  • func WriteTimeout(d time.Duration) PublisherOption
• type Subscriber
  • func NewSubscriber(config *Config, opts ...SubscriberOption) *Subscriber
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) GetEntity() *kafka.Reader
  • func (s *Subscriber) ReadMessage(ctx context.Context) (kafka.Message, error)
  • func (s *Subscriber) Subscribe(ctx context.Context, h Handler) error
• type SubscriberOption

Constants

const (
	// RequireNone the producer won’t even wait for a response from the broker.
	RequireNone kafka.RequiredAcks = kafka.RequireNone
	// RequireOne the producer will consider the write successful when the leader receives the record.
	RequireOne kafka.RequiredAcks = kafka.RequireOne
	// RequireAll the producer will consider the write successful when all of the in-sync replicas receive the record.
	RequireAll kafka.RequiredAcks = kafka.RequireAll
)

Variables

var ErrEventFull = errors.New("message event chan full")

ErrEventFull is a message event chan full.

Types

type Config

type Config struct {
	Brokers []string `yaml:"brokers"`
	Topic   string   `yaml:"topic"`
	GroupId string   `yaml:"groupId"`
}

type Event

type Event struct {
	// Key sets the key of the message for routing policy
	Key string
	// Payload for the message
	Payload []byte
	// Properties attach application defined properties on the message
	Properties map[string]string
	// topic
	Topic string
	// offset
	Offset int64
	// 分区
	Partition int
}

type Handler

type Handler func(context.Context, Event) error

type Publisher

type Publisher struct {
	// contains filtered or unexported fields
}

func NewPublisher

func NewPublisher(config *Config, opts ...PublisherOption) *Publisher

NewPublisher new a kafka publisher.

func (*Publisher) Close

func (p *Publisher) Close() error

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, event Event) error

type PublisherOption

type PublisherOption func(*Publisher)

PublisherOption is a publisher options.

func BatchSize

func BatchSize(size int) PublisherOption

func EventBuffer

func EventBuffer(n int) PublisherOption

EventBuffer with event buffer option.

func ReadTimeout

func ReadTimeout(d time.Duration) PublisherOption

ReadTimeout with read timeout option.

func RequiredAcks

func RequiredAcks(acks kafka.RequiredAcks) PublisherOption

RequiredAcks with required acks option.

func WriteTimeout

func WriteTimeout(d time.Duration) PublisherOption

WriteTimeout with write timeout option.

type Subscriber

type Subscriber struct {
	// contains filtered or unexported fields
}

func NewSubscriber

func NewSubscriber(config *Config, opts ...SubscriberOption) *Subscriber

NewSubscriber new a kafka subscriber.

func (*Subscriber) Close

func (s *Subscriber) Close() error

func (*Subscriber) GetEntity

func (s *Subscriber) GetEntity() *kafka.Reader

func (*Subscriber) ReadMessage

func (s *Subscriber) ReadMessage(ctx context.Context) (kafka.Message, error)

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, h Handler) error

type SubscriberOption

type SubscriberOption func(*Subscriber)

SubscriberOption is a subscriber option.