schedule

Documentation

Overview

Package schedule: Provides resource mapping and task scheduler

Index

• Variables
• type AsyncTask
• type AsyncTaskScheduler
  • func NewAsyncTaskScheduler(ctx context.Context, numWorker, chanSize int) *AsyncTaskScheduler
  • func (a *AsyncTaskScheduler) Close() error
  • func (a *AsyncTaskScheduler) Errors() <-chan error
  • func (a *AsyncTaskScheduler) Push(task AsyncTask) (err error)
  • func (a *AsyncTaskScheduler) Size() int32
• type ExponentialRetryConfig
• type ExponentialStrategy
  • func (r *ExponentialStrategy) Next(err error, n int) (retry bool, wait time.Duration)
• type ForeverRetryConfig
• type ForeverRetryStrategy
  • func (r *ForeverRetryStrategy) Next(err error, _ int) (retry bool, wait time.Duration)
• type MappedResource
• type MappedTask
• type MappedTaskManager
  • func NewTaskManager() *MappedTaskManager
  • func (t *MappedTaskManager) IsEmpty() bool
  • func (t *MappedTaskManager) PopRemainAndAddRun() (task MappedTask, ok bool)
  • func (t *MappedTaskManager) PushRemain(task MappedTask)
  • func (t *MappedTaskManager) RemoveRun(task MappedTask)
  • func (t *MappedTaskManager) RemoveRunAndPushRemain(task MappedTask)
  • func (t *MappedTaskManager) Runs() (tasks []MappedTask)
  • func (t *MappedTaskManager) Size() int
• type NTimesRetryConfig
• type NTimesRetryStrategy
  • func (r *NTimesRetryStrategy) Next(err error, n int) (retry bool, wait time.Duration)
• type NoneRetryStrategy
  • func (r *NoneRetryStrategy) Next(err error, n int) (retry bool, wait time.Duration)
• type Resource
• type ResourceMap
  • func NewResourceMap() *ResourceMap
  • func (r *ResourceMap) Get(key string, create func() (MappedResource, error)) (resource MappedResource, err error)
  • func (r *ResourceMap) Release(resource MappedResource) (err error)
  • func (r *ResourceMap) UseCount(resource MappedResource) int
• type RetryJudger
• type RetryStrategy
  • func NewExponentialRetryStrategy(j RetryJudger, init, max time.Duration) RetryStrategy
  • func NewForeverRetryStrategy(j RetryJudger, wait time.Duration) RetryStrategy
  • func NewNTimesRetryStrategy(j RetryJudger, n int, wait time.Duration) RetryStrategy
  • func NewNoneRetryStrategy() RetryStrategy
  • func NewRetryStrategy(j RetryJudger, conf *config.JSON) (s RetryStrategy, err error)
• type RetryTask
  • func NewRetryTask(ctx context.Context, strategy RetryStrategy, task Task) *RetryTask
  • func (r *RetryTask) Do() (err error)
• type Task
• type TaskSchduler
  • func NewTaskSchduler(workerNumer, cap int) *TaskSchduler
  • func (t *TaskSchduler) Push(task Task) (<-chan error, error)
  • func (t *TaskSchduler) Size() int32
  • func (t *TaskSchduler) Stop()

Variables

var (
	ErrClosed = errors.New("task schduler closed")
)

ErrorClosed - Represents an error indicating that the operation is closed.

Types

type AsyncTask

type AsyncTask interface {
	Do() error   // Synchronous execution
	Post() error // Follow-up notification
}

AsyncTask (Asynchronous Task)

type AsyncTaskScheduler

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

AsyncTaskScheduler: Asynchronous task scheduler

func NewAsyncTaskScheduler

func NewAsyncTaskScheduler(ctx context.Context,
	numWorker, chanSize int) *AsyncTaskScheduler

NewAsyncTaskScheduler: Create an asynchronous task scheduler using the context ctx, the number of parallel workers numWorker, and the task channel size chanSize. Create an asynchronous task scheduler with the specified context

func (*AsyncTaskScheduler) Close

func (a *AsyncTaskScheduler) Close() error

Close: Close the asynchronous task scheduler.

func (*AsyncTaskScheduler) Errors

func (a *AsyncTaskScheduler) Errors() <-chan error

Errors: Error listener for the asynchronous task scheduler.

func (*AsyncTaskScheduler) Push

func (a *AsyncTaskScheduler) Push(task AsyncTask) (err error)

Push: Asynchronously execute a task.

func (*AsyncTaskScheduler) Size

func (a *AsyncTaskScheduler) Size() int32

Size: The number of tasks currently in the asynchronous task scheduler.

type ExponentialRetryConfig

type ExponentialRetryConfig struct {
	Init time2.Duration `json:"init"`
	Max  time2.Duration `json:"max"`
}

ExponentialRetryConfig: Exponential backoff retry strategy

type ExponentialStrategy

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

ExponentialStrategy: Exponential backoff retry strategy

func (*ExponentialStrategy) Next

func (r *ExponentialStrategy) Next(err error, n int) (retry bool, wait time.Duration)

Next: Determine whether to retry the next attempt and the waiting time for the next attempt

type ForeverRetryConfig

type ForeverRetryConfig struct {
	Wait time2.Duration `json:"wait"`
}

ForeverRetryConfig: Permanent retry strategy

type ForeverRetryStrategy

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

ForeverRetryStrategy: Permanent retry strategy with no maximum attempt limit

func (*ForeverRetryStrategy) Next

func (r *ForeverRetryStrategy) Next(err error, _ int) (retry bool, wait time.Duration)

Next: Determine whether to retry the next attempt and the waiting time for the next attempt. In a permanent retry strategy

type MappedResource

type MappedResource interface {
	Resource

	Key() string // Keyword
}

MappedResource: Mappable resource

type MappedTask

type MappedTask interface {
	Key() string // Mapped primary key
}

MappedTask

type MappedTaskManager

type MappedTaskManager struct {
	sync.Mutex
	// contains filtered or unexported fields
}

MappedTaskManager task manager toto I don't know why len(remain) + len(run) can't accurately represent the number of real-time tasks, mainly because len(run) is not accurate

func NewTaskManager

func NewTaskManager() *MappedTaskManager

NewTaskManager create task manager

func (*MappedTaskManager) IsEmpty

func (t *MappedTaskManager) IsEmpty() bool

IsEmpty check if the task manager is empty

func (*MappedTaskManager) PopRemainAndAddRun

func (t *MappedTaskManager) PopRemainAndAddRun() (task MappedTask, ok bool)

PopRemainAndAddRun move task from pending queue to running queue

func (*MappedTaskManager) PushRemain

func (t *MappedTaskManager) PushRemain(task MappedTask)

PushRemain add task to pending queue

func (*MappedTaskManager) RemoveRun

func (t *MappedTaskManager) RemoveRun(task MappedTask)

RemoveRun remove task from running queue

func (*MappedTaskManager) RemoveRunAndPushRemain

func (t *MappedTaskManager) RemoveRunAndPushRemain(task MappedTask)

RemoveRunAndPushRemain move task from running queue to pending queue

func (*MappedTaskManager) Runs

func (t *MappedTaskManager) Runs() (tasks []MappedTask)

Runs get currently running tasks

func (*MappedTaskManager) Size

func (t *MappedTaskManager) Size() int

Size number of tasks

type NTimesRetryConfig

type NTimesRetryConfig struct {
	N    int            `json:"n"`
	Wait time2.Duration `json:"wait"`
}

NTimesRetryConfig: Retry strategy with a fixed number of attempts

type NTimesRetryStrategy

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

NTimesRetryStrategy: Retry strategy with a fixed number of attempts

func (*NTimesRetryStrategy) Next

func (r *NTimesRetryStrategy) Next(err error, n int) (retry bool, wait time.Duration)

Next: Determine whether to retry the next attempt and the waiting time for the next attempt

type NoneRetryStrategy

type NoneRetryStrategy struct{}

NoneRetryStrategy: No retry strategy

func (*NoneRetryStrategy) Next

func (r *NoneRetryStrategy) Next(err error, n int) (retry bool, wait time.Duration)

Next: Whether to retry the next attempt and the waiting time for the next attempt

type Resource

type Resource interface {
	Close() error // Close and release resources
}

Resource: Resource

type ResourceMap

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

ResourceMap: Resource mapping

func NewResourceMap

func NewResourceMap() *ResourceMap

NewResourceMap: Create a resource mapping

func (*ResourceMap) Get

func (r *ResourceMap) Get(key string, create func() (MappedResource, error)) (resource MappedResource, err error)

Get: Retrieve a resource based on the keyword key. If the resource does not exist If there is an error creating the resource

func (*ResourceMap) Release

func (r *ResourceMap) Release(resource MappedResource) (err error)

Release: Release a resource based on the resource itself. If the resource does not exist If there is an error creating the resource

func (*ResourceMap) UseCount

func (r *ResourceMap) UseCount(resource MappedResource) int

UseCount: Calculate the number of times a resource has been used based on the resource itself

type RetryJudger

type RetryJudger interface {
	ShouldRetry(err error) bool
}

RetryJudger: Retry decision-maker

type RetryStrategy

type RetryStrategy interface {
	Next(err error, n int) (retry bool, wait time.Duration)
}

RetryStrategy: Retry strategy interface or base class

func NewExponentialRetryStrategy

func NewExponentialRetryStrategy(j RetryJudger, init, max time.Duration) RetryStrategy

NewExponentialRetryStrategy: Create an exponential backoff retry strategy based on a retry judge

func NewForeverRetryStrategy

func NewForeverRetryStrategy(j RetryJudger, wait time.Duration) RetryStrategy

NewForeverRetryStrategy: Create a permanent retry strategy based on a retry judge and retry interval

func NewNTimesRetryStrategy

func NewNTimesRetryStrategy(j RetryJudger, n int, wait time.Duration) RetryStrategy

NewNTimesRetryStrategy: Create a retry strategy with a fixed number of attempts

func NewNoneRetryStrategy

func NewNoneRetryStrategy() RetryStrategy

NewNoneRetryStrategy: Create a strategy with no retries

func NewRetryStrategy

func NewRetryStrategy(j RetryJudger, conf *config.JSON) (s RetryStrategy, err error)

NewRetryStrategy: Generate a retry strategy based on the configuration file

type RetryTask

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

RetryTask retry task

func NewRetryTask

func NewRetryTask(ctx context.Context, strategy RetryStrategy, task Task) *RetryTask

NewRetryTask generates retry task based on context relationship ctx

func (*RetryTask) Do

func (r *RetryTask) Do() (err error)

Do synchronous execution

type Task

type Task interface {
	Do() error // Synchronous execution
}

Task

type TaskSchduler

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

TaskScheduler - Represents a task scheduler.

func NewTaskSchduler

func NewTaskSchduler(workerNumer, cap int) *TaskSchduler

NewTaskScheduler - Creates a new task scheduler based on the number of workers (workerNumber) and the capacity of the pending task queue (capacity).

func (*TaskSchduler) Push

func (t *TaskSchduler) Push(task Task) (<-chan error, error)

Push - Adds a task to the queue and receives a notification channel for the execution result. An error is reported if the queue is closed.

func (*TaskSchduler) Size

func (t *TaskSchduler) Size() int32

Size - Represents the size of the pending task queue.

func (*TaskSchduler) Stop

func (t *TaskSchduler) Stop()

Stop - Stops the task scheduler.