init

Author: 王哈哈

.gitignore

@@ -0,0 +1 @@
+.idea

README.md

@@ -0,0 +1,60 @@
+# Async Task
+
+![](asyncTask.jpg)
+
+安装
+```go
+go get github.com/wanghaha-dev/asyncTask
+```
+
+使用
+```go
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/wanghaha-dev/asyncTask/asyncTask"
+	"sync"
+)
+
+func main() {
+	ctx := context.Background()
+	task, err := asyncTask.NewTask(ctx, asyncTask.Config{
+		Addr:      "127.0.0.1:6379",
+		DB:       0,
+		Password: "",
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	// take task
+	go func() {
+		asyncTask.Each(1000, func() {
+			data, err := task.TakeNormalTask()
+			if err != nil {
+				panic(err)
+			}
+			fmt.Println("data:", data)
+		})
+		wg.Done()
+	}()
+
+	// put task
+	asyncTask.Each(1000, func() {
+		err = task.PutNormalTask("ddd", asyncTask.Map{
+			"name": "wanghaha",
+		})
+		if err != nil {
+			panic(err)
+		}
+	})
+
+	wg.Wait()
+	fmt.Println("finish.")
+}
+```

asyncTask.go

@@ -0,0 +1,174 @@
+package asyncTask
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/go-redis/redis/v8"
+	"time"
+)
+
+type Config struct {
+	Addr     string
+	DB       int
+	Password string
+}
+
+func NewTask(ctx context.Context, config Config) (*Task, error) {
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     config.Addr,
+		Password: config.Password, // no password set
+		DB:       config.DB,       // use default DB
+	})
+
+	task := &Task{
+		RedisClient: rdb,
+		ctx:         ctx,
+	}
+
+	return task, nil
+}
+
+type Map map[string]interface{}
+
+type Task struct {
+	RedisClient *redis.Client
+	ctx         context.Context
+}
+
+func (t *Task) putTask(taskId string, list string, data Map) error {
+	d := struct {
+		TaskId string `json:"task_id"`
+		Data   Map    `json:"data"`
+	}{
+		taskId,
+		data,
+	}
+
+	taskContent, err := json.Marshal(d)
+	if err != nil {
+		return err
+	}
+
+	err = t.RedisClient.LPush(t.ctx, list, taskContent).Err()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (t *Task) PutNormalTask(taskId string, data Map) error {
+	return t.putTask(taskId, "normal", data)
+}
+func (t *Task) PutSuccessTask(taskId string, data Map) error {
+	return t.putTask(taskId, "success", data)
+}
+func (t *Task) PutFailTask(taskId string, data Map) error {
+	return t.putTask(taskId, "fail", data)
+}
+
+func (t *Task) takeNormalTask(list string) (Map, error) {
+	sliceCmd := t.RedisClient.BRPop(t.ctx, 0, list)
+
+	err := sliceCmd.Err()
+	if err != nil {
+		return nil, err
+	}
+
+	result, err := sliceCmd.Result()
+	if err != nil {
+		return nil, err
+	}
+
+	data := make(map[string]interface{})
+	err = json.Unmarshal([]byte(result[1]), &data)
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+func (t *Task) TakeNormalTask() (Map, error) {
+	data, err := t.takeNormalTask("normal")
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+func (t *Task) TakeSuccessTask() (Map, error) {
+	data, err := t.takeNormalTask("success")
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+func (t *Task) TakeFailTask() (Map, error) {
+	data, err := t.takeNormalTask("fail")
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+func (t *Task) getTaskLength(list string) (int64, error) {
+	lLen := t.RedisClient.LLen(t.ctx, list)
+	err := lLen.Err()
+
+	if err != nil {
+		return -1, err
+	}
+
+	return lLen.Result()
+}
+
+func (t *Task) GetFailTaskLength() (int64, error) {
+	return t.getTaskLength("fail")
+}
+
+func (t *Task) GetSuccessTaskLength() (int64, error) {
+	return t.getTaskLength("success")
+}
+
+func (t *Task) GetNormalTaskLength() (int64, error) {
+	return t.getTaskLength("normal")
+}
+
+func (t *Task) getFailTaskList(list string) ([]string, error) {
+	sliceCmd := t.RedisClient.LRange(t.ctx, list, 0, -1)
+
+	err := sliceCmd.Err()
+	if err != nil {
+		return []string{}, err
+	}
+	return sliceCmd.Result()
+}
+
+func (t *Task) GetFailTaskList() ([]string, error) {
+	return t.getFailTaskList("fail")
+}
+
+func (t *Task) GetSuccessTaskList() ([]string, error) {
+	return t.getFailTaskList("success")
+}
+
+func (t *Task) GetNormalTaskList() ([]string, error) {
+	return t.getFailTaskList("normal")
+}
+
+func EachStrings(data []string) {
+	for _, item := range data {
+		fmt.Println(item)
+	}
+}
+
+func Each(count int, f func()) {
+	for i := 0; i < count; i++ {
+		f()
+	}
+}
+
+func Datetime() string {
+	return time.Now().Format("2006-01-02 15:04:05")
+}

asyncTask.jpg

@@ -0,0 +1,9 @@
+module github.com/wanghaha-dev/asyncTask
+
+go 1.17
+
+require (
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/go-redis/redis/v8 v8.11.5 // indirect
+)

go.mod

@@ -0,0 +1,14 @@
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
+github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
+github.com/gomodule/redigo v1.8.9 h1:Sl3u+2BI/kk+VEatbj0scLdrFhjPmbxOc1myhDP41ws=
+github.com/gomodule/redigo v1.8.9/go.mod h1:7ArFNvsTjH8GMMzB4uy1snslv2BwmginuMs06a1uzZE=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=