版权声明 本站原创文章 由 萌叔 发表
转载请注明 萌叔 | http://vearne.cc

起因

我们有一个业务场景(高并发)，收到的每个任务都必须在很短的时间内完成，然后回报给调用方，因此每个任务都需要设定定时器。显然golang 默认的timer是不能支持这种场景的，因此我在delayqueue的基础上自己实现了一个库gtimer
GitHub地址

timer

用golang实现的定时器，基于delayqueue

实现

实现受到了Java DelayQueue.java的启发 源码地址 DelayQueue.java

依赖的几个结构依次为为 timer -> delayqueue -> priorityqueue -> heap

由于golang的Condition不支持wait一段时间，所以使用golang原生的Timer来替代了Condition在delayqueue中的作用

Installation

Install:

go get -u github.com/vearne/gtimer

Import:

import "github.com/vearne/gtimer"

Quick Start

package main

import (
	"fmt"
	log "github.com/sirupsen/logrus"
	"github.com/vearne/gtimer"
	"math/rand"
	"strconv"
	"sync/atomic"
	"time"
)

const (
	PRODUCER_COUNT = 10
	CONSUMER_COUNT = 10
	TARGET_COUNT   = 1000000
)

var ops int64 = 0

func main() {
	st := gtimer.NewSuperTimer(CONSUMER_COUNT)
	t1 := time.Now()
	for i := 0; i < PRODUCER_COUNT; i++ {
		go push(st, "worker"+strconv.Itoa(i))
	}

	time.Sleep(100 * time.Millisecond)

	for {
		v := atomic.LoadInt64(&ops)
		if v >= TARGET_COUNT {
			st.Stop()
			break
		} else {
			time.Sleep(100 * time.Millisecond)
		}
	}
	t2 := time.Now()
	log.Infof("cost:%v\n", t2.Sub(t1))
}

func DefaultAction(t time.Time, value string) {
	// fmt.Printf("trigger_time:%v, value:%v\n", t, value)
	atomic.AddInt64(&ops, 1)
}

func push(timer *gtimer.SuperTimer, name string) {
	r := rand.New(rand.NewSource(time.Now().UnixNano()))
	for i := 0; i < 1000000; i++ {
		now := time.Now()
		t := now.Add(time.Millisecond * time.Duration(r.Int63n(300)))
		value := fmt.Sprintf("%v:value:%v", name, strconv.Itoa(i))
		// create a delayed task
		item := gtimer.NewDelayedItemFunc(t, value, DefaultAction)
		timer.Add(item)
	}
}

use NewDelayedItemFunc, we can create a task

// triggerTime is time of the task should be execute
func NewDelayedItemFunc(triggerTime time.Time, value string, f func(time.Time, string)) *Item

task struct like

type Item struct {
	value    string // The value of the item; arbitrary.
	priority int64    // The priority of the item in the queue.
	// The index is needed by update and is maintained by the heap.Interface methods.
	index int // The index of the item in the heap.
	// when task is ready, execute OnTrigger function
	OnTrigger func(time.Time, string)
}

Performance

CPU Model Name: 2.3 GHz Intel Core i5
CPU Processors: 4
Memory: 8GB

Benchmark Test Results

PS： 如果更多的考虑性能，推荐大家看看这篇文章 golang:一个高性能低精度timer实现 https://blog.csdn.net/siddontang/article/details/23826507

请我喝瓶饮料