1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
package main
import (
"time"
"sync"
"context"
"log"
)
// type Handler interface {
// DoTask(context.Context) error
// }
type Handler = func() error
type Task struct {
Id int64
Handler Handler
Err error
Result any
IsTimeout bool
Done chan struct{}
}
func (r *Task) Debug() {
if r.Result != nil || r.Err != nil {
log.Printf("任务完成 result %v error %v",r.Result,r.Err)
return
}
if r.IsTimeout {
log.Printf("任务超时退出 result %v error %v",r.Result,r.Err)
return
}
log.Printf("任务未完成 result %v error %v",r.Result,r.Err)
}
type TaskQueue struct {
Keys map[int64]*Task
Mu sync.Mutex
Done chan int64 //记录 完成任务的ID
}
func NewQueue(queueSize int64) (*TaskQueue) {
t := new(TaskQueue)
t.Keys = make(map[int64]*Task, queueSize)
t.Done = make(chan int64, queueSize)
return t
}
func (q *TaskQueue) RemoveId(id int64) *Task {
q.Mu.Lock()
defer q.Mu.Unlock()
t := q.Keys[id]
delete(q.Keys,id)
return t
}
func (q *TaskQueue) Len() int {
q.Mu.Lock()
defer q.Mu.Unlock()
return len(q.Keys)
}
func (q *TaskQueue) AddTask(c context.Context,r *Task,timeout time.Duration) error {
q.Mu.Lock()
defer q.Mu.Unlock()
q.Keys[r.Id] = r
go func (r *Task) {
go func () {
err := r.Handler()
// log.Printf("任务处理完成 %v",err)
r.Done<-struct{}{}
r.Err = err
close(r.Done)
}()
select {
case <-c.Done():
// context cancel
log.Printf("任务取消 %v",r.Id)
case <- r.Done:
// 处理完成
log.Printf("任务结束 %v",r.Id)
case <- time.After(timeout):
// 超时标记或者 返回
log.Printf("任务超时 %v",r.Id)
r.IsTimeout = true
}
q.Done <- r.Id
}(r)
return nil
}
var idx int64 = 1
func newTask(f func () error ) *Task {
t := new(Task)
t.Done = make(chan struct{},1)
t.Id = idx
idx++
t.Handler = func () error {
defer func () {
if t.Err != nil {
t.Result = "failed"
return
}
t.Result = "success"
}()
err := f()
t.Err = err
return err
}
return t
}
func main() {
q := NewQueue(2)
// 模拟任务处理
q.AddTask(context.TODO(), newTask(func () error {
time.Sleep(time.Second * 25)
log.Printf("任务阻塞异常")
return nil
}),time.Second * 20)
q.AddTask(context.TODO(), newTask(func () error {
time.Sleep(time.Second * 10)
return nil
}),time.Second * 20)
var wg sync.WaitGroup
wg.Add(2)
if q.Len() == 2 {
// 假设任务数量为2，等待队列的任务处理完成再加
for q.Len() > 0 {
successId := <- q.Done
doneTask := q.RemoveId(successId)
doneTask.Debug()
wg.Done()
}
}
wg.Wait()
log.Printf("结束")
}
|
