first commit

Author: Vic Shóstak

.editorconfig

@@ -0,0 +1,16 @@
+root = true
+
+[*]
+indent_style = space
+indent_size = 2
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true
+
+[{go.mod,go.sum,*.go}]
+indent_style = tab
+indent_size = 4
+
+[{Makefile,Dockerfile,*.yml,*.yaml}]
+indent_style = tab
+indent_size = 2

.gitattributes

@@ -0,0 +1 @@
+go.sum merge=union

.gitignore

@@ -0,0 +1,12 @@
+# macOS
+**/.DS_Store
+
+# Dev build
+build/
+tmp/
+
+# Test
+*.out
+
+# Environment variables
+.env

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2021 Vic Shóstak <truewebartisans@gmail.com> (https://1wa.co)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Makefile

@@ -0,0 +1,7 @@
+.PHONY: worker client
+
+worker: 
+	go run ./worker/server.go
+
+stop: 
+	go run ./client/main.go

README.md

@@ -0,0 +1,28 @@
+# 📖 Tutorial: Asynq. Simple, reliable & efficient distributed task queue for your next Go project
+
+👉 The full article is published on **April 7, 2021**, on Dev.to:
+
+## Quick start
+
+1. Clone this repository and go to it root folder.
+2. Start a Redis server (by Docker or locally).
+3. Start Asynq worker server:
+
+```console
+make worker
+```
+
+4. Start generating tasks by Asynq client:
+
+```console
+make client
+```
+
+5. Install [Asynqmon](https://github.com/hibiken/asynqmon) (Asynq web UI) to your system.
+6. Go to [localhost:8080](http://localhost:8080) and see:
+
+![Screenshot](https://user-images.githubusercontent.com/11155743/113557216-57af2b80-9606-11eb-8ab6-df023b14e5c1.png)
+
+## ⚠️ License
+
+MIT © [Vic Shóstak](https://shostak.dev/) & [True web artisans](https://1wa.co/).

client/main.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"log"
+	"math/rand"
+	"time"
+
+	"tutorial-go-asynq/tasks"
+
+	"github.com/hibiken/asynq"
+)
+
+func main() {
+	// Create a new Redis connection for the client.
+	redisConnection := asynq.RedisClientOpt{
+		Addr: "localhost:6379", // Redis server address
+	}
+
+	// Create a new Asynq client.
+	client := asynq.NewClient(redisConnection)
+	defer client.Close()
+
+	// Infinite loop to create tasks as Asynq client.
+	for {
+		// Generate a random user ID.
+		userID := rand.Intn(1000) + 10
+
+		// Set a delay duration to 2 minutes.
+		delay := 2 * time.Minute
+
+		// Define tasks.
+		task1 := tasks.NewWelcomeEmailTask(userID)
+		task2 := tasks.NewReminderEmailTask(userID, time.Now().Add(delay))
+
+		// Process the task immediately in critical queue.
+		if _, err := client.Enqueue(
+			task1,                   // task payload
+			asynq.Queue("critical"), // set queue for task
+		); err != nil {
+			log.Fatal(err)
+		}
+
+		// Process the task 2 minutes later in low queue.
+		if _, err := client.Enqueue(
+			task2,                  // task payload
+			asynq.Queue("low"),     // set queue for task
+			asynq.ProcessIn(delay), // set time to process task
+		); err != nil {
+			log.Fatal(err)
+		}
+	}
+}

go.mod

@@ -0,0 +1,5 @@
+module tutorial-go-asynq
+
+go 1.16
+
+require github.com/hibiken/asynq v0.17.1

go.sum

@@ -0,0 +1,61 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/go-redis/redis/v7 v7.4.0 h1:7obg6wUoj05T0EpY0o8B59S9w5yeMWql7sw2kwNW1x4=
+github.com/go-redis/redis/v7 v7.4.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
+github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/hibiken/asynq v0.17.1 h1:vNnK+wbnzjzs9E6WN0Im8M6a6oxXaRKK5wue1KsEC30=
+github.com/hibiken/asynq v0.17.1/go.mod h1:yfQUmjFqSBSUIVxTK0WyW4LPj4gpr283UpWb6hKYaqE=
+github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
+github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.10.1 h1:q/mM8GF/n0shIN8SaAZ0V+jnLPzen6WIVZdiwrRlMlo=
+github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/gomega v1.7.0 h1:XPnZz8VVBHjVsy1vzJmRwIcSwiUO+JFfrv/xGiigmME=
+github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+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/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
+github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng=
+github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
+github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+go.uber.org/goleak v0.10.0 h1:G3eWbSNIskeRqtsN/1uI5B+eP73y3JUuBsv9AZjehb4=
+go.uber.org/goleak v0.10.0/go.mod h1:VCZuO8V8mFPlL0F5J5GK1rtHV3DrFcQ1R8ryq7FK0aI=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478 h1:l5EDrHhldLYb3ZRHDUhXF7Om7MvYXnkV9/iQNo1lX6g=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e h1:9vRrk9YW2BTzLP0VCB9ZDjU4cPqkg+IDWL7XgxA1yxQ=
+golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
+golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
+gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
+gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.7 h1:VUgggvou5XRW9mHwD/yXxIYSMtY0zoKQf/v226p2nyo=
+gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

tasks/handlers.go

@@ -0,0 +1,43 @@
+package tasks
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hibiken/asynq"
+)
+
+// HandleWelcomeEmailTask handler for welcome email task.
+func HandleWelcomeEmailTask(c context.Context, t *asynq.Task) error {
+	// Get user ID from given task.
+	id, err := t.Payload.GetInt("user_id")
+	if err != nil {
+		return err
+	}
+
+	// Dummy message to the worker's output.
+	fmt.Printf("Send Welcome Email to User ID %d\n", id)
+
+	return nil
+}
+
+// HandleReminderEmailTask for reminder email task.
+func HandleReminderEmailTask(c context.Context, t *asynq.Task) error {
+	// Get int with the user ID from the given task.
+	id, err := t.Payload.GetInt("user_id")
+	if err != nil {
+		return err
+	}
+
+	// Get string with the sent time from the given task.
+	time, err := t.Payload.GetString("sent_in")
+	if err != nil {
+		return err
+	}
+
+	// Dummy message to the worker's output.
+	fmt.Printf("Send Reminder Email to User ID %d\n", id)
+	fmt.Printf("Reason: time is up (%v)\n", time)
+
+	return nil
+}

tasks/payloads.go

@@ -0,0 +1,40 @@
+package tasks
+
+import (
+	"time"
+
+	"github.com/hibiken/asynq"
+)
+
+const (
+	// TypeWelcomeEmail is a name of the task type
+	// for sending a welcome email.
+	TypeWelcomeEmail = "email:welcome"
+
+	// TypeReminderEmail is a name of the task type
+	// for sending a reminder email.
+	TypeReminderEmail = "email:reminder"
+)
+
+// NewWelcomeEmailTask task payload for a new welcome email.
+func NewWelcomeEmailTask(id int) *asynq.Task {
+	// Specify task payload.
+	payload := map[string]interface{}{
+		"user_id": id, // set user ID
+	}
+
+	// Return a new task with given type and payload.
+	return asynq.NewTask(TypeWelcomeEmail, payload)
+}
+
+// NewReminderEmailTask task payload for a reminder email.
+func NewReminderEmailTask(id int, ts time.Time) *asynq.Task {
+	// Specify task payload.
+	payload := map[string]interface{}{
+		"user_id": id,          // set user ID
+		"sent_in": ts.String(), // set time to sending
+	}
+
+	// Return a new task with given type and payload.
+	return asynq.NewTask(TypeReminderEmail, payload)
+}

worker/server.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"log"
+
+	"tutorial-go-asynq/tasks"
+
+	"github.com/hibiken/asynq"
+)
+
+func main() {
+	// Create and configuring Redis connection.
+	redisConnection := asynq.RedisClientOpt{
+		Addr: "localhost:6379", // Redis server address
+	}
+
+	// Create and configuring Asynq worker server.
+	worker := asynq.NewServer(redisConnection, asynq.Config{
+		// Specify how many concurrent workers to use.
+		Concurrency: 10,
+		// Specify multiple queues with different priority.
+		Queues: map[string]int{
+			"critical": 6, // processed 60% of the time
+			"default":  3, // processed 30% of the time
+			"low":      1, // processed 10% of the time
+		},
+	})
+
+	// Create a new task's mux instance.
+	mux := asynq.NewServeMux()
+
+	// Define a task handler for the welcome email task.
+	mux.HandleFunc(
+		tasks.TypeWelcomeEmail,       // task type
+		tasks.HandleWelcomeEmailTask, // handler function
+	)
+
+	// Define a task handler for the reminder email task.
+	mux.HandleFunc(
+		tasks.TypeReminderEmail,       // task type
+		tasks.HandleReminderEmailTask, // handler function
+	)
+
+	// Run worker server.
+	if err := worker.Run(mux); err != nil {
+		log.Fatal(err)
+	}
+}