init

task/init.go

@@ -0,0 +1,9 @@
+package task
+
+func Init() {
+	NewTaskListen().Listen()
+
+	//NewTask()("order", 60, NewOrder()).Push()
+
+	NewTaskQueue().Queue()
+}

task/listen.go

@@ -0,0 +1,59 @@
+package task
+
+import (
+	"Edu/config"
+	"Edu/serve/cache"
+	"Edu/serve/cache/logic"
+	"Edu/utils"
+	"fmt"
+	"sync"
+)
+
+type TaskListen struct {
+	Produce chan *Task
+	Consume chan *Task
+	lock    *sync.Mutex
+}
+
+var TaskListenEvent *TaskListen
+
+func (this *TaskListen) Join(task *Task) {
+	TaskListenEvent.Produce <- task
+}
+
+func (this *TaskListen) Quit(task *Task) {
+	TaskListenEvent.Consume <- task
+}
+
+func (this *TaskListen) Listen() {
+	go utils.TryCatch(func() {
+		for {
+			select {
+			case p := <-this.Produce:
+				err := cache.Cache.ZAdd(config.RedisKeyForTaskQueue, &logic.ScoreParams{Score: float64(p.DelayUnix()), Member: p.ID})
+
+				if err != nil {
+					fmt.Printf("Task Produce Redis Sadd Error【%s】", err)
+				} else {
+					err = cache.Cache.HSet(config.RedisKeyForTaskQueueBody, p.ID, p)
+					fmt.Printf("err：%v\n", err)
+				}
+			case c := <-this.Consume:
+				if err := c.Handle(); err != nil {
+					fmt.Printf("err：%v\n", err)
+				}
+			}
+		}
+	})
+}
+
+func NewTaskListen() *TaskListen {
+	if TaskListenEvent == nil {
+		TaskListenEvent = &TaskListen{
+			Produce: make(chan *Task, 1),
+			Consume: make(chan *Task, 1),
+			lock:    new(sync.Mutex),
+		}
+	}
+	return TaskListenEvent
+}

task/order.go

@@ -0,0 +1,27 @@
+package task
+
+import (
+	"Edu/utils"
+	"encoding/json"
+)
+
+type Order struct {
+	ID   string `json:"id"`
+	Name string `json:"name"`
+}
+
+func (this *Order) MarshalBinary() ([]byte, error) {
+	return json.Marshal(this)
+}
+
+func (this *Order) UnmarshalBinary(data []byte) error {
+	return utils.FromJSONBytes(data, this)
+}
+
+func (this *Order) Handle() error {
+	return nil
+}
+
+func NewOrder() *Order {
+	return &Order{ID: "23", Name: "Henry"}
+}

task/queue.go

@@ -0,0 +1,41 @@
+package task
+
+import (
+	"Edu/config"
+	cache2 "Edu/serve/cache"
+	"Edu/serve/cache/logic"
+	"Edu/utils"
+	"fmt"
+	"time"
+)
+
+type TaskQueue struct{}
+
+func (this *TaskQueue) Queue() {
+	go utils.TryCatch(func() {
+		for {
+			now := time.Now()
+
+			cache, _ := cache2.Cache.ZRangebyscore(config.RedisKeyForTaskQueue, &logic.ScoreRangeBy{Min: "0",
+				Max: fmt.Sprintf("%d", now.Unix())})
+
+			if len(cache) > 0 {
+				for _, v := range cache {
+					body, _ := cache2.Cache.HGet(config.RedisKeyForTaskQueueBody, v)
+					task := new(Task)
+					_ = task.UnmarshalBinary([]byte(body))
+					task.Consume()
+				}
+				// 销毁信息
+				_ = cache2.Cache.ZRem(config.RedisKeyForTaskQueue, cache)
+				_ = cache2.Cache.HDel(config.RedisKeyForTaskQueueBody, cache...)
+			}
+			// 每秒执行一次
+			time.Sleep(time.Second)
+		}
+	})
+}
+
+func NewTaskQueue() *TaskQueue {
+	return &TaskQueue{}
+}

task/task.go

@@ -0,0 +1,62 @@
+package task
+
+import (
+	"Edu/utils"
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+type ITask interface {
+	MarshalBinary() ([]byte, error)
+	UnmarshalBinary(data []byte) error
+	Handle() error
+}
+
+type Task struct {
+	ID    string `json:"id"`    // Job唯一标识
+	Topic string `json:"topic"` // Job类型
+	Delay int64  `json:"delay"` // Job需要延迟的时间, 单位：秒
+	Body  ITask  `json:"body"`  // Job的内容，供消费者做具体的业务处理
+}
+
+type TaskHandle func(topic string, delay int64, body ITask) *Task
+
+func (this *Task) MarshalBinary() ([]byte, error) {
+	return json.Marshal(this)
+}
+
+func (this *Task) UnmarshalBinary(data []byte) error {
+	return utils.FromJSONBytes(data, this)
+}
+
+func (this *Task) setID() string {
+	return utils.Sha1String(fmt.Sprintf("%s%d%s", this.Topic, time.Now().UnixNano(), utils.GetRandomCode(6)))
+}
+
+func (this *Task) DelayUnix() int64 {
+	return time.Now().Unix() + this.Delay
+}
+
+func (this *Task) Push() {
+	TaskListenEvent.Join(this)
+}
+
+// Consume 消费
+func (this *Task) Consume() {
+	TaskListenEvent.Quit(this)
+}
+
+func (this *Task) Handle() error {
+	// 处理各种方法
+	this.Body.(*Order).Handle()
+	return nil
+}
+
+func NewTask() TaskHandle {
+	return func(topic string, delay int64, body ITask) *Task {
+		task := &Task{Topic: topic, Delay: delay, Body: body}
+		task.ID = task.setID()
+		return task
+	}
+}