docs: add NSQ example (#10)

Signed-off-by: Bo-Yi Wu <[email protected]>

Author: appleboy

README.md

@@ -37,12 +37,38 @@ Second to create the new worker, use buffered channel as example:
 	w := simple.NewWorker(
 		simple.WithQueueNum(taskN),
 		simple.WithRunFunc(func(m queue.QueuedMessage) error {
-			j, ok := m.(*job)
+			v, ok := m.(*job)
 			if !ok {
-				return errors.New("message is not job type")
+				if err := json.Unmarshal(m.Bytes(), &v); err != nil {
+					return err
+				}
 			}
 
-			rets <- j.message
+			rets <- v.Message
+			return nil
+		}),
+	)
+```
+
+or you can use the [NSQ](https://nsq.io/) as backend, see the worker example:
+
+```go
+	// define the worker
+	w := nsq.NewWorker(
+		nsq.WithAddr("127.0.0.1:4150"),
+		nsq.WithTopic("example"),
+		nsq.WithChannel("foobar"),
+		// concurrent job number
+		nsq.WithMaxInFlight(10),
+		nsq.WithRunFunc(func(m queue.QueuedMessage) error {
+			v, ok := m.(*job)
+			if !ok {
+				if err := json.Unmarshal(m.Bytes(), &v); err != nil {
+					return err
+				}
+			}
+
+			rets <- v.Message
 			return nil
 		}),
 	)

_example/nsq/go.mod

@@ -0,0 +1,5 @@
+module example
+
+go 1.16
+
+require github.com/appleboy/queue v0.0.4-0.20210725052905-4ef008157945

_example/nsq/go.sum

@@ -0,0 +1,18 @@
+github.com/appleboy/queue v0.0.3 h1:rntqVTm6ilh80VCVQjwA0vDMCl1cfveq6GS6X98fKwE=
+github.com/appleboy/queue v0.0.3/go.mod h1:6Mn0z4hURZW/26huvRXG0SJ4o7pBdo6hOryRiegy/4Q=
+github.com/appleboy/queue v0.0.4-0.20210725052905-4ef008157945 h1:Mec41QJ0hGDtbyfWgHfuXKqTu0a8hSIIBhL3sNbNjqQ=
+github.com/appleboy/queue v0.0.4-0.20210725052905-4ef008157945/go.mod h1:6Mn0z4hURZW/26huvRXG0SJ4o7pBdo6hOryRiegy/4Q=
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/nsqio/go-nsq v1.0.8 h1:3L2F8tNLlwXXlp2slDUrUWSBn2O3nMh8R1/KEDFTHPk=
+github.com/nsqio/go-nsq v1.0.8/go.mod h1:vKq36oyeVXgsS5Q8YEO7WghqidAVXQlcFxzQbQTuDEY=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+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 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+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 h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

_example/nsq/main.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/appleboy/queue"
+	"github.com/appleboy/queue/nsq"
+)
+
+type job struct {
+	Message string
+}
+
+func (j *job) Bytes() []byte {
+	b, err := json.Marshal(j)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
+func main() {
+	taskN := 100
+	rets := make(chan string, taskN)
+
+	// define the worker
+	w := nsq.NewWorker(
+		nsq.WithAddr("127.0.0.1:4150"),
+		nsq.WithTopic("example"),
+		nsq.WithChannel("foobar"),
+		// concurrent job number
+		nsq.WithMaxInFlight(10),
+		nsq.WithRunFunc(func(m queue.QueuedMessage) error {
+			v, ok := m.(*job)
+			if !ok {
+				if err := json.Unmarshal(m.Bytes(), &v); err != nil {
+					return err
+				}
+			}
+
+			rets <- v.Message
+			return nil
+		}),
+	)
+
+	// define the queue
+	q, err := queue.NewQueue(
+		queue.WithWorkerCount(10),
+		queue.WithWorker(w),
+	)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// start the five worker
+	q.Start()
+
+	// assign tasks in queue
+	for i := 0; i < taskN; i++ {
+		go func(i int) {
+			q.Queue(&job{
+				Message: fmt.Sprintf("handle the job: %d", i+1),
+			})
+		}(i)
+	}
+
+	// wait until all tasks done
+	for i := 0; i < taskN; i++ {
+		fmt.Println("message:", <-rets)
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	q.Shutdown()
+	q.Wait()
+}

_example/simple/main.go

@@ -1,7 +1,7 @@
 package main
 
 import (
-	"errors"
+	"encoding/json"
 	"fmt"
 	"log"
 	"time"
@@ -11,11 +11,11 @@ import (
 )
 
 type job struct {
-	message string
+	Message string
 }
 
 func (j *job) Bytes() []byte {
-	return []byte(j.message)
+	return []byte(j.Message)
 }
 
 func main() {
@@ -26,12 +26,14 @@ func main() {
 	w := simple.NewWorker(
 		simple.WithQueueNum(taskN),
 		simple.WithRunFunc(func(m queue.QueuedMessage) error {
-			j, ok := m.(*job)
+			v, ok := m.(*job)
 			if !ok {
-				return errors.New("message is not job type")
+				if err := json.Unmarshal(m.Bytes(), &v); err != nil {
+					return err
+				}
 			}
 
-			rets <- j.message
+			rets <- v.Message
 			return nil
 		}),
 	)
@@ -52,7 +54,7 @@ func main() {
 	for i := 0; i < taskN; i++ {
 		go func(i int) {
 			q.Queue(&job{
-				message: fmt.Sprintf("handle the job: %d", i+1),
+				Message: fmt.Sprintf("handle the job: %d", i+1),
 			})
 		}(i)
 	}