queue

package
v0.0.0-...-e43f524 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 16, 2026 License: AGPL-3.0 Imports: 14 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// ErrQueueClosed 队列已关闭
	ErrQueueClosed = 20001
	// ErrTopicClosed Topic 已关闭
	ErrTopicClosed = 20002
	// ErrChannelClosed Channel 已关闭
	ErrChannelClosed = 20003
	// ErrChannelNotFound Channel 不存在
	ErrChannelNotFound = 20004
	// ErrNoMessage 没有可用消息
	ErrNoMessage = 20005
)

错误码定义

Variables

This section is empty.

Functions

func GenerateMessageID 

func GenerateMessageID() string

GenerateMessageID 生成消息唯一标识

func NewKafkaWriter 

func NewKafkaWriter(brokers []string, topic string, config *KafkaConfig) *kafka.Writer

NewKafkaWriter 创建 Kafka Writer

Types

type Channel 

type Channel[T any] interface {
	// Name 返回 Channel 名称
	Name() string

	// Subscribe 订阅消息，使用回调函数处理
	Subscribe(handler Handler[T])

	// Next 获取下一条消息（阻塞）
	Next() (*Message[T], error)

	// TryNext 尝试获取下一条消息（可设置超时）
	TryNext(timeout time.Duration) (*Message[T], error)

	// Ack 确认消息已成功处理
	Ack(msgId string) error

	// Nack 消息处理失败，重新入队
	Nack(msgId string) error

	// Depth 返回 Channel 深度（包括待处理和未确认）
	Depth() (int64, error)

	// Close 关闭 Channel
	Close() error
}

Channel Channel 接口，负责消息的消费

type ChannelConfig 

type ChannelConfig struct {
	// MaxRetries 最大重试次数
	MaxRetries int `json:"max_retries,omitempty" yaml:"max_retries,omitempty" toml:"max_retries,omitempty" default:"5"`
	// RetryDelay 重试延迟
	RetryDelay time.Duration `json:"retry_delay,omitempty" yaml:"retry_delay,omitempty" toml:"retry_delay,omitempty" default:"1s"`
	// MessageTTL 消息过期时间
	MessageTTL time.Duration `json:"message_ttl,omitempty" yaml:"message_ttl,omitempty" toml:"message_ttl,omitempty" default:"24h"`
	// MaxInFlight 最大飞行消息数（同时消费的消息数）
	MaxInFlight int `json:"max_in_flight,omitempty" yaml:"max_in_flight,omitempty" toml:"max_in_flight,omitempty" default:"10"`
	// AckTimeout 超时自动重新入队
	AckTimeout time.Duration `json:"ack_timeout,omitempty" yaml:"ack_timeout,omitempty" toml:"ack_timeout,omitempty" default:"30s"`
}

ChannelConfig Channel 配置

type Config 

type Config struct {
	// StorageType 存储类型
	StorageType StorageType `json:"storage_type,omitempty" yaml:"storage_type,omitempty" toml:"storage_type,omitempty" default:"memory"`
	// MaxRetries 最大重试次数
	MaxRetries int `json:"max_retries,omitempty" yaml:"max_retries,omitempty" toml:"max_retries,omitempty" default:"5"`
	// RetryDelay 重试延迟
	RetryDelay time.Duration `json:"retry_delay,omitempty" yaml:"retry_delay,omitempty" toml:"retry_delay,omitempty" default:"1s"`
	// MessageTTL 消息过期时间
	MessageTTL time.Duration `json:"message_ttl,omitempty" yaml:"message_ttl,omitempty" toml:"message_ttl,omitempty" default:"24h"`
	// MaxBodySize 最大消息体大小
	MaxBodySize int64 `json:"max_body_size,omitempty" yaml:"max_body_size,omitempty" toml:"max_body_size,omitempty" default:"1048576"`
	// MaxMsgSize 最大消息数量
	MaxMsgSize int64 `json:"max_msg_size,omitempty" yaml:"max_msg_size,omitempty" toml:"max_msg_size,omitempty" default:"1000000"`

	// RedisConfig Redis 配置（当 StorageType 为 redis 时使用）
	RedisConfig *RedisConfig `json:"redis_config,omitempty" yaml:"redis_config,omitempty" toml:"redis_config,omitempty"`

	// RedisClient 外部传入的 Redis 客户端（优先级高于 RedisConfig）
	RedisClient *redis.Client `json:"-" yaml:"-" toml:"-"`

	// KafkaConfig Kafka 配置（当 StorageType 为 kafka 时使用）
	KafkaConfig *KafkaConfig `json:"kafka_config,omitempty" yaml:"kafka_config,omitempty" toml:"kafka_config,omitempty"`
}

Config 队列配置

type Handler 

type Handler[T any] func(msg *Message[T]) (ProcessRsp, error)

Handler 消息处理函数类型

type KafkaConfig 

type KafkaConfig struct {
	// Brokers Kafka 服务器地址列表（格式：host:port）
	Brokers []string `json:"brokers,omitempty" yaml:"brokers,omitempty" toml:"brokers,omitempty"`
	// TopicPrefix Topic 名称前缀
	TopicPrefix string `json:"topic_prefix,omitempty" yaml:"topic_prefix,omitempty" toml:"topic_prefix,omitempty" default:"lrpc-queue-"`
	// Partition 使用的分区数（创建 Topic 时使用）
	Partition int `json:"partition,omitempty" yaml:"partition,omitempty" toml:"partition,omitempty" default:"1"`
	// ReplicationFactor 副本因子（创建 Topic 时使用）
	ReplicationFactor int `json:"replication_factor,omitempty" yaml:"replication_factor,omitempty" toml:"replication_factor,omitempty" default:"1"`
	// ConsumerGroupID 消费者组 ID
	ConsumerGroupID string `json:"consumer_group_id,omitempty" yaml:"consumer_group_id,omitempty" toml:"consumer_group_id,omitempty"`
	// AutoCreateTopics 是否自动创建 Topic
	AutoCreateTopics bool `json:"auto_create_topics,omitempty" yaml:"auto_create_topics,omitempty" toml:"auto_create_topics,omitempty" default:"true"`
	// ReadBatchTimeout 批量读取超时
	ReadBatchTimeout time.Duration `json:"read_batch_timeout,omitempty" yaml:"read_batch_timeout,omitempty" toml:"read_batch_timeout,omitempty" default:"10s"`
	// WriteTimeout 写入超时
	WriteTimeout time.Duration `json:"write_timeout,omitempty" yaml:"write_timeout,omitempty" toml:"write_timeout,omitempty" default:"10s"`
	// RequiredAcks 确认级别（0=无需确认，1=leader确认，-1=all确认）
	RequiredAcks int `json:"required_acks,omitempty" yaml:"required_acks,omitempty" toml:"required_acks,omitempty" default:"1"`
	// CompressionType 压缩类型（none, gzip, snappy, lz4, zstd）
	CompressionType string `json:"compression_type,omitempty" yaml:"compression_type,omitempty" toml:"compression_type,omitempty" default:"none"`
	// SessionTimeout 会话超时
	SessionTimeout time.Duration `json:"session_timeout,omitempty" yaml:"session_timeout,omitempty" toml:"session_timeout,omitempty" default:"30s"`
	// RebalanceTimeout 重平衡超时
	RebalanceTimeout time.Duration `json:"rebalance_timeout,omitempty" yaml:"rebalance_timeout,omitempty" toml:"rebalance_timeout,omitempty" default:"60s"`
	// CommitInterval 提交间隔
	CommitInterval time.Duration `json:"commit_interval,omitempty" yaml:"commit_interval,omitempty" toml:"commit_interval,omitempty" default:"1s"`
	// HeartbeatInterval 心跳间隔
	HeartbeatInterval time.Duration `json:"heartbeat_interval,omitempty" yaml:"heartbeat_interval,omitempty" toml:"heartbeat_interval,omitempty" default:"3s"`
	// MaxAttempts 最大消费尝试次数
	MaxAttempts int `json:"max_attempts,omitempty" yaml:"max_attempts,omitempty" toml:"max_attempts,omitempty" default:"5"`
	// DialTimeout 连接超时
	DialTimeout time.Duration `json:"dial_timeout,omitempty" yaml:"dial_timeout,omitempty" toml:"dial_timeout,omitempty" default:"10s"`
}

KafkaConfig Kafka 配置

type Message 

type Message[T any] struct {
	// Id 消息唯一标识
	Id string `json:"id,omitempty" yaml:"id,omitempty" toml:"id,omitempty"`
	// Body 消息体，使用泛型支持任意类型
	Body T `json:"body,omitempty" yaml:"body,omitempty" toml:"body,omitempty"`
	// Timestamp 消息产生时间戳
	Timestamp int64 `json:"timestamp,omitempty" yaml:"timestamp,omitempty" toml:"timestamp,omitempty"`
	// ExpiresAt 消息过期时间戳，0 表示永不过期
	ExpiresAt int64 `json:"expires_at,omitempty" yaml:"expires_at,omitempty" toml:"expires_at,omitempty"`
	// Attempts 消费尝试次数
	Attempts int `json:"attempts,omitempty" yaml:"attempts,omitempty" toml:"attempts,omitempty"`

	// Channel 所属 Channel
	Channel string `json:"channel,omitempty" yaml:"channel,omitempty" toml:"channel,omitempty"`
}

Message 泛型消息结构

func NewMessage 

func NewMessage[T any](body T) *Message[T]

NewMessage 创建新消息

func NewMessageWithID 

func NewMessageWithID[T any](id string, body T) *Message[T]

NewMessageWithID 使用指定 ID 创建消息

func (*Message[T]) Clone 

func (m *Message[T]) Clone() *Message[T]

Clone 创建消息的副本（用于不同 Channel）

func (*Message[T]) GetTTL 

func (m *Message[T]) GetTTL() int64

GetTTL 获取消息剩余存活时间 返回剩余时间（秒），-1 表示永不过期，0 表示已过期

func (*Message[T]) IncrementAttempts 

func (m *Message[T]) IncrementAttempts()

IncrementAttempts 增加尝试次数

func (*Message[T]) IsExpired 

func (m *Message[T]) IsExpired() bool

IsExpired 检查消息是否已过期 返回 true 表示已过期，false 表示未过期或永不过期

func (*Message[T]) ResetAttempts 

func (m *Message[T]) ResetAttempts()

ResetAttempts 重置尝试次数

func (*Message[T]) SetExpires 

func (m *Message[T]) SetExpires(duration time.Duration)

SetExpires 设置消息过期时间 duration 为相对时间（如 30 * time.Minute），0 表示永不过期

func (*Message[T]) SetExpiresAt 

func (m *Message[T]) SetExpiresAt(timestamp int64)

SetExpiresAt 设置消息过期时间戳 timestamp 为绝对时间戳（Unix 秒），0 表示永不过期

type ProcessRsp 

type ProcessRsp struct {
	// Retry 是否需要重试
	Retry bool
	// SkipAttempts 跳过记录重试次数
	SkipAttempts bool
}

ProcessRsp 消息处理响应

func Handle 

func Handle[T any](handler Handler[T], msg *Message[T]) (rsp ProcessRsp, err error)

type Queue 

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

func NewQueue 

func NewQueue(c *Config) *Queue

func (*Queue) Close 

func (q *Queue) Close() error

Close 关闭队列

type RedisConfig 

type RedisConfig struct {
	// Addr Redis 服务器地址（格式：host:port）
	Addr string `json:"addr,omitempty" yaml:"addr,omitempty" toml:"addr,omitempty" default:"localhost:6379" env:"QUEUE_ADDR"`
	// Password Redis 密码
	Password string `json:"password,omitempty" yaml:"password,omitempty" toml:"password,omitempty" env:"QUEUE_PASSWORD"`
	// DB Redis 数据库编号
	DB int `json:"db,omitempty" yaml:"db,omitempty" toml:"db,omitempty" default:"0" env:"QUEUE_DB"`
	// KeyPrefix Redis 键名前缀
	KeyPrefix string `` /* 128-byte string literal not displayed */
	// PoolSize 连接池大小
	PoolSize int `json:"pool_size,omitempty" yaml:"pool_size,omitempty" toml:"pool_size,omitempty" default:"10" env:"QUEUE_POOL_SIZE"`
	// MinIdleConns 最小空闲连接数
	MinIdleConns int `` /* 134-byte string literal not displayed */
	// MaxRetries 最大重试次数
	MaxRetries int `json:"max_retries,omitempty" yaml:"max_retries,omitempty" toml:"max_retries,omitempty" default:"3" env:"QUEUE_MAX_RETRIES"`
	// DialTimeout 连接超时
	DialTimeout time.Duration `` /* 127-byte string literal not displayed */
	// ReadTimeout 读取超时
	ReadTimeout time.Duration `` /* 127-byte string literal not displayed */
	// WriteTimeout 写入超时
	WriteTimeout time.Duration `` /* 131-byte string literal not displayed */
	// PoolTimeout 连接池超时
	PoolTimeout time.Duration `` /* 127-byte string literal not displayed */
}

RedisConfig Redis 配置

type StorageType 

type StorageType string
const (
	StorageMemory StorageType = "memory"
	StorageRedis  StorageType = "redis"
	StorageKafka  StorageType = "kafka"
)

type Topic 

type Topic[T any] interface {
	// Pub 发布消息到 Topic
	Pub(msg T) error

	// PubBatch 批量发布消息
	PubBatch(msgs []T) error

	// PubMsg 发布完整消息（包含元数据）
	PubMsg(msg *Message[T]) error

	// PubMsgBatch 批量发布完整消息
	PubMsgBatch(msgs []*Message[T]) error

	// GetOrAddChannel 获取或创建一个 Channel
	GetOrAddChannel(name string, config *ChannelConfig) (Channel[T], error)

	// GetChannel 获取已存在的 Channel
	GetChannel(name string) (Channel[T], error)

	// ChannelList 返回所有 Channel 名称
	ChannelList() []string

	// Close 关闭 Topic
	Close() error
}

Topic Topic 接口，负责消息的生产和分发

func NewKafkaTopic 

func NewKafkaTopic[T any](writer *kafka.Writer, name string, config *TopicConfig, prefix string, brokers []string, kafkaConfig *KafkaConfig) Topic[T]

NewKafkaTopic 创建 Kafka Topic

func NewMemoryTopic 

func NewMemoryTopic[T any](name string, config *TopicConfig) Topic[T]

NewMemoryTopic 创建内存 Topic

func NewRedisTopic 

func NewRedisTopic[T any](cli *redis.Client, name string, config *TopicConfig, prefix string) Topic[T]

NewRedisTopic 创建 Redis Topic

func NewTopic 

func NewTopic[T any](queue *Queue, name string, topic *TopicConfig) Topic[T]

type TopicConfig 

type TopicConfig struct {
	// MaxRetries 最大重试次数
	MaxRetries int `json:"max_retries,omitempty" yaml:"max_retries,omitempty" toml:"max_retries,omitempty" default:"5"`
	// RetryDelay 重试延迟
	RetryDelay time.Duration `json:"retry_delay,omitempty" yaml:"retry_delay,omitempty" toml:"retry_delay,omitempty" default:"1s"`
	// MessageTTL 消息过期时间
	MessageTTL time.Duration `json:"message_ttl,omitempty" yaml:"message_ttl,omitempty" toml:"message_ttl,omitempty" default:"24h"`
	// MaxBodySize 最大消息体大小
	MaxBodySize int64 `json:"max_body_size,omitempty" yaml:"max_body_size,omitempty" toml:"max_body_size,omitempty" default:"1048576"`
	// MaxMsgSize 最大消息数量
	MaxMsgSize int64 `json:"max_msg_size,omitempty" yaml:"max_msg_size,omitempty" toml:"max_msg_size,omitempty" default:"1000000"`
}

TopicConfig Topic 配置

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.