kafka

Documentation

Index

• func InstallMiddleware(handle ...ProducerHandler)
• type Consumer
  • func GetConsumer() Consumer
• type ConsumerConfig
• type ConsumerImpl
  • func (c *ConsumerImpl) Booting(bootManager freedom.BootManager)
  • func (c *ConsumerImpl) Close() error
  • func (c *ConsumerImpl) Restart() error
  • func (c *ConsumerImpl) Start(config *ConsumerConfig)
• type Msg
  • func (msg *Msg) GetExecution() error
  • func (msg *Msg) GetHeader() map[string]interface{}
  • func (msg *Msg) GetMessageKey() string
  • func (msg *Msg) IsStopped() bool
  • func (msg *Msg) Next()
  • func (msg *Msg) Publish() error
  • func (msg *Msg) SetHeader(head map[string]interface{}) *Msg
  • func (msg *Msg) SetMessageKey(key string) *Msg
  • func (msg *Msg) Stop() *Msg
• type Producer
  • func GetProducer() Producer
• type ProducerHandler
• type ProducerImpl
  • func (pi *ProducerImpl) Booting(bootManager freedom.BootManager)
  • func (pi *ProducerImpl) Close() error
  • func (pi *ProducerImpl) NewMsg(topic string, content []byte) *Msg
  • func (pi *ProducerImpl) Restart() error
  • func (pi *ProducerImpl) Start(addrs []string, config *sarama.Config)
• type WorkerPool
  • func NewWorkerPool(maxWorkers int) *WorkerPool
  • func (p *WorkerPool) Close(timeout time.Duration) error
  • func (p *WorkerPool) Submit(task func())

Functions

func InstallMiddleware

func InstallMiddleware(handle ...ProducerHandler)

InstallMiddleware Install the middleware.. You can control the publishing of messages by installing middleware.

Types

type Consumer

type Consumer interface {
	// Start 使用配置结构体启动消费者
	Start(config *ConsumerConfig)
	// Restart the connection.
	Restart() error
	// Close the connection.
	Close() error
}

Consumer Kafka Consumer interface definition.

func GetConsumer

func GetConsumer() Consumer

GetConsumer Returns the Consumer instance.

type ConsumerConfig

type ConsumerConfig struct {
	// Kafka 服务器地址
	Addrs []string
	// 消费者组ID
	GroupID string
	// Sarama 配置
	Config *sarama.Config
	// 代理地址
	ProxyAddr string
	// 是否使用 H2C 代理
	ProxyH2C bool
	// HTTP 请求超时时间
	RequestTimeout time.Duration
	// 并发限制（并行模式下最大同时处理的消息数，默认500）
	RateLimit int
	// 优雅关闭超时时间（默认3秒）
	CloseTimeout time.Duration
}

ConsumerConfig 消费者配置结构体

type ConsumerImpl

type ConsumerImpl struct {
	freedom.Infra
	// contains filtered or unexported fields
}

ConsumerImpl Kafka Consumer implementation.

func (*ConsumerImpl) Booting

func (c *ConsumerImpl) Booting(bootManager freedom.BootManager)

Booting The method of overriding the component . The single-case component initiates a callback.

func (*ConsumerImpl) Close

func (c *ConsumerImpl) Close() error

Close the connection.

func (*ConsumerImpl) Restart

func (c *ConsumerImpl) Restart() error

Restart the connection.

func (*ConsumerImpl) Start

func (c *ConsumerImpl) Start(config *ConsumerConfig)

Start

type Msg

type Msg struct {
	Topic string

	Content []byte
	// contains filtered or unexported fields
}

Msg Kafka Message.

func (*Msg) GetExecution

func (msg *Msg) GetExecution() error

GetExecution Get the results of the execution .

func (*Msg) GetHeader

func (msg *Msg) GetHeader() map[string]interface{}

GetHeader .

func (*Msg) GetMessageKey

func (msg *Msg) GetMessageKey() string

GetMessageKey Get kafka key.

func (*Msg) IsStopped

func (msg *Msg) IsStopped() bool

IsStopped whether it has stopped.

func (*Msg) Next

func (msg *Msg) Next()

Next Perform the next step, typically for the control of middleware.

func (*Msg) Publish

func (msg *Msg) Publish() error

Publish this message.

func (*Msg) SetHeader

func (msg *Msg) SetHeader(head map[string]interface{}) *Msg

SetHeader set up HTTP Header.

func (*Msg) SetMessageKey

func (msg *Msg) SetMessageKey(key string) *Msg

SetMessageKey Set kafka key.

func (*Msg) Stop

func (msg *Msg) Stop() *Msg

Stop execution, typically used for middleware control

type Producer

type Producer interface {
	// Create a new message
	NewMsg(topic string, content []byte) *Msg
	// Start pass in the relevant address, configuration.
	Start(addrs []string, config *sarama.Config)
	// Restart the connection.
	Restart() error
}

Producer The producer's interface definition.

func GetProducer

func GetProducer() Producer

GetProducer Gets an instance of the producer.

type ProducerHandler

type ProducerHandler func(*Msg)

ProducerHandler The function declaration of the Kafka Producer middleware..

type ProducerImpl

type ProducerImpl struct {
	freedom.Infra
	// contains filtered or unexported fields
}

ProducerImpl The realization of the producer.

func (*ProducerImpl) Booting

func (pi *ProducerImpl) Booting(bootManager freedom.BootManager)

Booting The method of overriding the component . The single-case component initiates a callback.

func (*ProducerImpl) Close

func (pi *ProducerImpl) Close() error

Close .

func (*ProducerImpl) NewMsg

func (pi *ProducerImpl) NewMsg(topic string, content []byte) *Msg

NewMsg Create a new message.

func (*ProducerImpl) Restart

func (pi *ProducerImpl) Restart() error

Restart the connection.

func (*ProducerImpl) Start

func (pi *ProducerImpl) Start(addrs []string, config *sarama.Config)

Start pass in the relevant address, configuration.

type WorkerPool

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

WorkerPool

func NewWorkerPool

func NewWorkerPool(maxWorkers int) *WorkerPool

NewWorkerPool

func (*WorkerPool) Close

func (p *WorkerPool) Close(timeout time.Duration) error

Close

func (*WorkerPool) Submit

func (p *WorkerPool) Submit(task func())

Submit