Skip to content

Commit

Permalink
Feature/task redis server v7 (#13)
Browse files Browse the repository at this point in the history
* Add StopWithTimeout method.

* Upgrade task redis lib dependency: from v8 to v9.
  • Loading branch information
zhoufeng1989 authored Nov 8, 2023
1 parent 096b4c8 commit fae6595
Show file tree
Hide file tree
Showing 3 changed files with 40 additions and 2 deletions.
1 change: 1 addition & 0 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ require (
github.com/go-playground/validator/v10 v10.9.0
github.com/go-redis/redis/v8 v8.4.3
github.com/panjf2000/ants/v2 v2.4.3
github.com/redis/go-redis/v9 v9.3.0
github.com/relvacode/iso8601 v1.1.0
github.com/spf13/pflag v1.0.3
github.com/stretchr/testify v1.7.0
Expand Down
9 changes: 8 additions & 1 deletion go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,13 @@ github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+Ce
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b/go.mod h1:ac9efd0D1fsDb3EJvhqgXRbFx7bs2wqZ10HQPeU8U/Q=
github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
github.com/danieljoos/wincred v1.0.2 h1:zf4bhty2iLuwgjgpraD2E9UbvO+fe54XXGJbOwe23fU=
github.com/danieljoos/wincred v1.0.2/go.mod h1:SnuYRW9lp1oJrZX/dXJqr0cPK5gYXqx3EJbmjhLdK9U=
Expand Down Expand Up @@ -223,6 +228,8 @@ github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R
github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
github.com/prometheus/procfs v0.1.3 h1:F0+tqvhOksq22sc6iCHF5WGlWjdwj92p0udFh1VFBS8=
github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
github.com/redis/go-redis/v9 v9.3.0 h1:RiVDjmig62jIWp7Kk4XVLs0hzV6pI3PyTnnL0cnn0u0=
github.com/redis/go-redis/v9 v9.3.0/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
github.com/relvacode/iso8601 v1.1.0 h1:2nV8sp0eOjpoKQ2vD3xSDygsjAx37NHG2UlZiCkDH4I=
github.com/relvacode/iso8601 v1.1.0/go.mod h1:FlNp+jz+TXpyRqgmM7tnzHHzBnz776kmAH2h3sZCn0I=
github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
Expand Down
32 changes: 31 additions & 1 deletion task/task.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@ import (
"sync/atomic"
"time"

"github.com/go-redis/redis/v8"
"github.com/panjf2000/ants/v2"
"github.com/redis/go-redis/v9"
)

const (
Expand Down Expand Up @@ -194,6 +194,36 @@ func (scheduler *Scheduler) Stop(force bool) {
}
}

// StopWithTimeout stops the current scheduler, waiting running tasks at most `timeout` duration.
func (scheduler *Scheduler) StopWithTimeout(timeout time.Duration) {
if !atomic.CompareAndSwapInt32(&scheduler.started, 1, 0) {
return
}
scheduler.stop <- true
if timeout <= 0 {
scheduler.workerPool.Release()
return
}
timer := time.NewTimer(timeout)
defer timer.Stop()
for {
select {
case <-timer.C:
scheduler.workerPool.Release()
return
default:
runningCount := scheduler.runningJobCount()
if runningCount > 0 {
log.Printf("waiting %d jobs to finish...\n", runningCount)
} else {
scheduler.workerPool.Release()
return
}
time.Sleep(10 * time.Millisecond)
}
}
}

// JobStats returns all jobs' statistics in the current scheduler.
func (scheduler *Scheduler) JobStats() map[string][]JobStat {
jobStats := make(map[string][]JobStat, len(scheduler.jobs))
Expand Down

0 comments on commit fae6595

Please sign in to comment.