timer

README

timer

Go implementation of Kafka's Hierarchical Timing Wheels, Support millions of tasks. timed is a global timer instance, that tick is 1ms. wheel size is 1024, use ants goroutine pool.

Usage

Installation

Use go get.

    go get github.com/thinkgos/timer

Then import the package into your own code.

    import "github.com/thinkgos/timer"

Example

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/thinkgos/timer/timed"
)

func main() {
	var wg sync.WaitGroup
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		index := i
		_, _ = timed.AfterFunc(time.Duration(i)*100*time.Millisecond, func() {
			fmt.Printf("%s: timer task %d is executed, remain task: %d\n", time.Now().String(), index, timed.TaskCounter())
			wg.Done()
		})
	}
	wg.Wait()
}

References

• kafka timing wheels

License

This project is under MIT License. See the LICENSE file for the full license text.

Documentation

Index

• Constants
• Variables
• func IsPowOf2(x int) bool
• func NextPowOf2(x int) int
• type GoPool
• type Job
• type JobFunc
  • func (f JobFunc) Run()
• type Option
  • func WithGoPool(p GoPool) Option
  • func WithTickMs(tickMs int64) Option
  • func WithWheelSize(size int) Option
• type Spoke
  • func NewSpoke(taskCounter *atomic.Int64) *Spoke
  • func (sp *Spoke) Add(task *Task)
  • func (sp *Spoke) Delay() int64
  • func (sp *Spoke) Flush(f func(*Task))
  • func (sp *Spoke) GetExpiration() int64
  • func (sp *Spoke) Remove(task *Task)
  • func (sp *Spoke) SetExpiration(expirationMs int64) bool
• type Task
  • func NewTask(d time.Duration) *Task
  • func NewTaskFunc(d time.Duration, f func()) *Task
  • func (t *Task) Cancel()
  • func (t *Task) Cancelled() bool
  • func (t *Task) Delay() time.Duration
  • func (t *Task) ExpirationMs() int64
  • func (t *Task) Run()
  • func (t *Task) WithJob(j Job) *Task
  • func (t *Task) WithJobFunc(f func()) *Task
• type Timer
  • func NewTimer(opts ...Option) *Timer
  • func (t *Timer) AddTask(task *Task) error
  • func (t *Timer) AfterFunc(d time.Duration, f func()) (*Task, error)
  • func (t *Timer) Start()
  • func (t *Timer) Started() bool
  • func (t *Timer) Stop()
  • func (t *Timer) TaskCounter() int64
  • func (t *Timer) TickMs() int64
  • func (t *Timer) WheelMask() int
  • func (t *Timer) WheelSize() int
• type TimingWheel

Examples

• Timer

Constants

const (
	// DefaultTickMs default tick milliseconds.
	DefaultTickMs = 1
	// DefaultWheelSize default wheel size.
	DefaultWheelSize = 1024
	// DefaultTimeUnit default time unit is milliseconds.
	DefaultTimeUnit = time.Millisecond
)

Variables

var (
	// ErrClosed is returned when the timer is closed.
	ErrClosed = errors.New("timer: use of closed timer")
)

Functions

func IsPowOf2

func IsPowOf2(x int) bool

func NextPowOf2

func NextPowOf2(x int) int

Types

type GoPool

type GoPool interface {
	Go(f func())
}

GoPool goroutine pool.

type Job

type Job interface {
	Run()
}

Job job interface

type JobFunc

type JobFunc func()

JobFunc job function

func (JobFunc) Run

func (f JobFunc) Run()

Run implement job interface

type Option

type Option func(*Timer)

Option `Timer` custom options.

func WithGoPool

func WithGoPool(p GoPool) Option

WithGoPool set goroutine pool.

func WithTickMs

func WithTickMs(tickMs int64) Option

WithTickMs set basic time tick milliseconds.

func WithWheelSize

func WithWheelSize(size int) Option

WithWheelSize set wheel size.

type Spoke

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

Spoke a spoke of the wheel.

func NewSpoke

func NewSpoke(taskCounter *atomic.Int64) *Spoke

func (*Spoke) Add

func (sp *Spoke) Add(task *Task)

Add the timer task to this list

func (*Spoke) Delay

func (sp *Spoke) Delay() int64

Delay implements delayqueue.Delayed.

func (*Spoke) Flush

func (sp *Spoke) Flush(f func(*Task))

Flush all task entries and apply the supplied function to each of them

func (*Spoke) GetExpiration

func (sp *Spoke) GetExpiration() int64

Get the spoke's expiration time

func (*Spoke) Remove

func (sp *Spoke) Remove(task *Task)

Remove the specified timer task from this list

func (*Spoke) SetExpiration

func (sp *Spoke) SetExpiration(expirationMs int64) bool

Set the spoke's expiration time Returns true if the expiration time is changed

type Task

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

Task 是双向链表的一个元素.

func NewTask

func NewTask(d time.Duration) *Task

NewTask new task with delay duration and empty job, the accuracy is milliseconds.

func NewTaskFunc

func NewTaskFunc(d time.Duration, f func()) *Task

NewTaskFunc new task with delay duration and function job, the accuracy is milliseconds.

func (*Task) Cancel

func (t *Task) Cancel()

Cancel the task

func (*Task) Cancelled

func (t *Task) Cancelled() bool

Cancelled return true if the task is cancelled.

func (*Task) Delay

func (t *Task) Delay() time.Duration

Delay delay duration, the accuracy is milliseconds.

func (*Task) ExpirationMs

func (t *Task) ExpirationMs() int64

ExpirationMs expiration milliseconds.

func (*Task) Run

func (t *Task) Run()

Run immediate call job.

func (*Task) WithJob

func (t *Task) WithJob(j Job) *Task

WithJob with job

func (*Task) WithJobFunc

func (t *Task) WithJobFunc(f func()) *Task

WithJobFunc with function job

type Timer

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

Timer is a timer

Example

tm := NewTimer()
tm.Start()
_, _ = tm.AfterFunc(100*time.Millisecond, func() {
	fmt.Println(100)
})
_ = tm.AddTask(NewTask(1025 * time.Millisecond).WithJobFunc(func() {
	fmt.Println(200)
}))
canceledTaskAfterAdd := NewTask(300 * time.Millisecond).WithJobFunc(func() {
	fmt.Println("canceled after add")
})
_ = tm.AddTask(canceledTaskAfterAdd)
canceledTaskAfterAdd.Cancel()
canceledTaskBeforeAdd := NewTask(301 * time.Millisecond).WithJobFunc(func() {
	fmt.Println("canceled before add")
})
canceledTaskBeforeAdd.Cancel()
_ = tm.AddTask(canceledTaskBeforeAdd)
time.Sleep(time.Second + time.Millisecond*200)
tm.Stop()

Output:

100
200

func NewTimer

func NewTimer(opts ...Option) *Timer

NewTimer new timer instance. tick is 1 milliseconds, wheel size is 1024.

func (*Timer) AddTask

func (t *Timer) AddTask(task *Task) error

AddTask adds a task to the timer.

func (*Timer) AfterFunc

func (t *Timer) AfterFunc(d time.Duration, f func()) (*Task, error)

AfterFunc adds a function to the timer.

func (*Timer) Start

func (t *Timer) Start()

Start the timer.

func (*Timer) Started

func (t *Timer) Started() bool

Started have started or not.

func (*Timer) Stop

func (t *Timer) Stop()

Stop the timer.

func (*Timer) TaskCounter

func (t *Timer) TaskCounter() int64

TaskCounter the total number of tasks.

func (*Timer) TickMs

func (t *Timer) TickMs() int64

TickMs basic time tick milliseconds.

func (*Timer) WheelMask

func (t *Timer) WheelMask() int

WheelMask wheel mask.

func (*Timer) WheelSize

func (t *Timer) WheelSize() int

WheelSize wheel size.

type TimingWheel

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