workerpool

Documentation

Index

• Variables
• type GlobalTaskDispatcher
  • func NewGlobalTaskDispatcher(rdb redis.Cmdable, queues []string) *GlobalTaskDispatcher
  • func (d *GlobalTaskDispatcher) FindAndClaimTask(ctx context.Context, workerID string) (string, string, string, error)
  • func (d *GlobalTaskDispatcher) HasPendingTask(ctx context.Context) (bool, error)
• type Worker
  • func NewWorker(rdb redis.Cmdable, dispatcher *GlobalTaskDispatcher, ...) *Worker
• type WorkerPool
  • func NewWorkerPool(rdb redis.Cmdable, registry *consumer.HandlerRegistry, rm account.Manager, ...) *WorkerPool
  • func (p *WorkerPool) Start(ctx context.Context) error
  • func (p *WorkerPool) Stop(ctx context.Context) error

Variables

var WorkerPoolProviderSet = wire.NewSet(NewWorkerPool)

Types

type GlobalTaskDispatcher

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

GlobalTaskDispatcher 会扫描多个待处理队列，并以原子方式将任务移入特定工作线程的进行中队列。

func NewGlobalTaskDispatcher

func NewGlobalTaskDispatcher(rdb redis.Cmdable, queues []string) *GlobalTaskDispatcher

NewGlobalTaskDispatcher 创建一个分发器，使用给定的 Redis 客户端和队列名称。

func (*GlobalTaskDispatcher) FindAndClaimTask

func (d *GlobalTaskDispatcher) FindAndClaimTask(ctx context.Context, workerID string) (string, string, string, error)

FindAndClaimTask 尝试从任何待处理队列获取任务。 它返回原始任务消息、基础队列名称和生成的进行中队列名称。

func (*GlobalTaskDispatcher) HasPendingTask

func (d *GlobalTaskDispatcher) HasPendingTask(ctx context.Context) (bool, error)

HasPendingTask 检查是否有任何待处理队列包含任务。

type Worker

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

Worker 是一个绑定到账号的单个执行单元。

func NewWorker

func NewWorker(rdb redis.Cmdable, dispatcher *GlobalTaskDispatcher, registry *consumer.HandlerRegistry, rm account.Manager, sm *strategy.Manager, woRepo data.WorkOrderRepo, conf *conf.Data, logger log.Logger) *Worker

NewWorker 创建一个工人实例。

type WorkerPool

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

WorkerPool 管理工人池的生命周期。

func NewWorkerPool

func NewWorkerPool(rdb redis.Cmdable, registry *consumer.HandlerRegistry, rm account.Manager, woRepo data.WorkOrderRepo, conf *conf.Data, logger log.Logger) *WorkerPool

NewWorkerPool 创建一个工人池，所有依赖项均已连接。

func (*WorkerPool) Start

func (p *WorkerPool) Start(ctx context.Context) error

Start 启动工人池。

func (*WorkerPool) Stop

func (p *WorkerPool) Stop(ctx context.Context) error

Stop 停止工人池。