database

Documentation

Index

• func FiltrWhere(field field.Expr, value any) field.Expr
• func QueryPaging(query any, page Paging) error
• func SetDsn(m *configs.GormDbConfig) string
• type Aggregator
  • func NewAggregator(batchProcessor BatchProcessFunc, optionFuncs ...SetAggregatorOptionFunc) (*Aggregator, error)
  • func (agt *Aggregator) Enqueue(item interface{}) error
  • func (agt *Aggregator) EnqueueWithRetry(item interface{}, maxRetries int, backoff time.Duration) bool
  • func (agt *Aggregator) SafeStop()
  • func (agt *Aggregator) Start()
  • func (agt *Aggregator) Stop()
  • func (agt *Aggregator) TryEnqueue(item interface{}) bool
• type AggregatorOption
• type BatchProcessFunc
• type DBConfig
• type DBManager
  • func GetDBManager() *DBManager
  • func GetGormDB(dbConfig *configs.GormDbConfig) (dbm *DBManager, err error)
  • func (m *DBManager) GetDB() *gorm.DB
  • func (m *DBManager) InitDB(config DBConfig) error
  • func (m *DBManager) ReconnectIfNeeded() error
  • func (m *DBManager) StartHealthCheck(interval time.Duration)
• type ErrorHandlerFunc
• type Paging
  • func New() Paging
• type SetAggregatorOptionFunc
  • func WithBatchSize(batchSize int) SetAggregatorOptionFunc
  • func WithChannelBufferSize(size int) SetAggregatorOptionFunc
  • func WithErrorHandler(handler ErrorHandlerFunc) SetAggregatorOptionFunc
  • func WithLingerTime(duration time.Duration) SetAggregatorOptionFunc
  • func WithLogger(logger *log.Logger) SetAggregatorOptionFunc
  • func WithWorkers(workers int) SetAggregatorOptionFunc

Functions

func FiltrWhere

func FiltrWhere(field field.Expr, value any) field.Expr

func QueryPaging

func QueryPaging(query any, page Paging) error

func SetDsn

func SetDsn(m *configs.GormDbConfig) string

Types

type Aggregator

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

Aggregator 聚合器结构体

func NewAggregator

func NewAggregator(batchProcessor BatchProcessFunc, optionFuncs ...SetAggregatorOptionFunc) (*Aggregator, error)

NewAggregator 创建新的聚合器实例

func (*Aggregator) Enqueue

func (agt *Aggregator) Enqueue(item interface{}) error

Enqueue 入队一个项目，会阻塞直到有空间

func (*Aggregator) EnqueueWithRetry

func (agt *Aggregator) EnqueueWithRetry(item interface{}, maxRetries int, backoff time.Duration) bool

EnqueueWithRetry 入队一个项目，会重试直到成功或者达到最大重试次数

func (*Aggregator) SafeStop

func (agt *Aggregator) SafeStop()

SafeStop 安全停止聚合器，确保所有项目都被处理

func (*Aggregator) Start

func (agt *Aggregator) Start()

Start 启动聚合器

func (*Aggregator) Stop

func (agt *Aggregator) Stop()

Stop 停止聚合器

func (*Aggregator) TryEnqueue

func (agt *Aggregator) TryEnqueue(item interface{}) bool

TryEnqueue 尝试入队一个项目，非阻塞

type AggregatorOption

type AggregatorOption struct {
	BatchSize         int
	Workers           int
	ChannelBufferSize int
	LingerTime        time.Duration
	ErrorHandler      ErrorHandlerFunc
	Logger            *log.Logger
}

AggregatorOption 聚合器选项

type BatchProcessFunc

type BatchProcessFunc func([]interface{}) error

BatchProcessFunc 批处理函数类型

type DBConfig

type DBConfig struct {
	Driver string
	DSN    string
	Config *gorm.Config
}

DBConfig 数据库配置结构体

type DBManager

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

DBManager 数据库管理器

func GetDBManager

func GetDBManager() *DBManager

GetDBManager 获取数据库管理器单例

func GetGormDB

func GetGormDB(dbConfig *configs.GormDbConfig) (dbm *DBManager, err error)

GetGormDB 获取gorm.DB

func (*DBManager) GetDB

func (m *DBManager) GetDB() *gorm.DB

GetDB 获取数据库连接

func (*DBManager) InitDB

func (m *DBManager) InitDB(config DBConfig) error

InitDB 初始化数据库连接

func (*DBManager) ReconnectIfNeeded

func (m *DBManager) ReconnectIfNeeded() error

ReconnectIfNeeded 检查连接并在需要时重连

func (*DBManager) StartHealthCheck

func (m *DBManager) StartHealthCheck(interval time.Duration)

StartHealthCheck 启动健康检查

type ErrorHandlerFunc

type ErrorHandlerFunc func(err error, items []interface{}, batchProcessFunc BatchProcessFunc, aggregator *Aggregator)

ErrorHandlerFunc 错误处理函数类型

type Paging

type Paging interface {
	OrderBy() string
	SetTotal(int)
	GetTotal() int
	OffsetLimit() (offset, limit int)
}

func New

func New() Paging

type SetAggregatorOptionFunc

type SetAggregatorOptionFunc func(option *AggregatorOption)

SetAggregatorOptionFunc 聚合器选项设置函数类型

func WithBatchSize

func WithBatchSize(batchSize int) SetAggregatorOptionFunc

示例: 设置聚合器选项的函数

func WithChannelBufferSize

func WithChannelBufferSize(size int) SetAggregatorOptionFunc

func WithErrorHandler

func WithErrorHandler(handler ErrorHandlerFunc) SetAggregatorOptionFunc

func WithLingerTime

func WithLingerTime(duration time.Duration) SetAggregatorOptionFunc

func WithLogger

func WithLogger(logger *log.Logger) SetAggregatorOptionFunc

func WithWorkers

func WithWorkers(workers int) SetAggregatorOptionFunc