kafka

Documentation

Overview

Package kafka 提供 Kafka 消息队列集成(对标 Spring Kafka 的 KafkaTemplate): 生产/消费封装 + 原生客户端暴露。基于纯 Go 的 segmentio/kafka-go。

Index

• func SetConsumer(consumer *Consumer)
• func SetProducer(producer *Producer)
• type Config
• type Consumer
  • func GetConsumer() *Consumer
  • func NewConsumer(config Config) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) Consume(ctx context.Context, handler Handler) error
  • func (c *Consumer) Reader() *kafka.Reader
• type Handler
• type Message
• type Producer
  • func GetProducer() *Producer
  • func NewProducer(config Config) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) Send(ctx context.Context, topic, key string, value []byte) error
  • func (p *Producer) SendJSON(ctx context.Context, topic, key string, value interface{}) error
  • func (p *Producer) Writer() *kafka.Writer

Functions

func SetConsumer

func SetConsumer(consumer *Consumer)

SetConsumer 设置全局消费者(装配后调用)。

func SetProducer

func SetProducer(producer *Producer)

SetProducer 设置全局生产者(装配后调用)。

Types

type Config

type Config struct {
	// Brokers 节点列表(如 127.0.0.1:9092)。
	Brokers []string
	// Topic 默认主题(可空,操作时显式指定)。
	Topic string
	// GroupID 消费组(Consumer 使用)。
	GroupID string
	// Timeout 网络超时(默认 10s)。
	Timeout time.Duration
}

Config Kafka 客户端配置。

type Consumer

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

Consumer 消费端(对标 @KafkaListener)。

func GetConsumer

func GetConsumer() *Consumer

GetConsumer 获取全局消费者;未注册返回 nil。

func NewConsumer

func NewConsumer(config Config) (*Consumer, error)

NewConsumer 创建消费者(同一消费组内分区均衡)。

func (*Consumer) Close

func (c *Consumer) Close() error

Close 关闭消费者。

func (*Consumer) Consume

func (c *Consumer) Consume(ctx context.Context, handler Handler) error

Consume 阻塞消费:循环拉取消息并调用 handler;ctx 取消时返回 nil。 handler 返回 nil 后自动提交偏移(At-Least-Once 语义)。

func (*Consumer) Reader

func (c *Consumer) Reader() *kafka.Reader

Reader 返回原生 Reader(高级操作入口)。

type Handler

type Handler func(ctx context.Context, message Message) error

Handler 消息处理函数;返回 error 时消息被标记失败(不提交偏移)。

type Message

type Message struct {
	Topic     string
	Key       string
	Value     []byte
	Partition int
	Offset    int64
	Time      time.Time
	Headers   map[string]string
}

Message 消费消息。

type Producer

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

Producer 生产端(对标 KafkaTemplate.send)。

func GetProducer

func GetProducer() *Producer

GetProducer 获取全局生产者;未注册返回 nil。

func NewProducer

func NewProducer(config Config) (*Producer, error)

NewProducer 创建生产者。

func (*Producer) Close

func (p *Producer) Close() error

Close 关闭生产者(刷盘)。

func (*Producer) Send

func (p *Producer) Send(ctx context.Context, topic, key string, value []byte) error

Send 发送消息(key 为 nil 时按哈希均衡)。

func (*Producer) SendJSON

func (p *Producer) SendJSON(ctx context.Context, topic, key string, value interface{}) error

SendJSON 发送 JSON 消息。

func (*Producer) Writer

func (p *Producer) Writer() *kafka.Writer

Writer 返回原生 Writer(高级操作入口)。