Skip to content

Instantly share code, notes, and snippets.

@Ryouku

Ryouku/golang_job_queue.md

Forked from harlow/golang_job_queue.md
Created January 8, 2017 14:26
Show Gist options
  • Save Ryouku/f15e643382eef45efb83e1695ad3e764 to your computer and use it in GitHub Desktop.
Save Ryouku/f15e643382eef45efb83e1695ad3e764 to your computer and use it in GitHub Desktop.
Download ZIP
Job queues in Golang
Raw
golang_job_queue.md

Golang Job Queue

A running example of the code from:

Step 1

Small refactorings made to original code:

  • Use non-exported private methods
  • Remove global variables
  • Bring the flags closer to their usage in main()

Step 2

Simplify the worker queue by removing the Dispatcher.

  • Creates workers directly and passes job queue to them

https://gist.github.com/harlow/dbcd639cf8d396a2ab73#file-worker_refactored-go

Performance

The test run with Pprof show performance characteristics remain the same between both examples.

Run the Application

Boot either the worker_original.go or the worker_refactored.go applications. Use flags to adjust the max_workers and max_queue_size to override the default values.

$ go run worker_original.go -max_workers 5

cURL the application from another terminal window:

$ for i in {1..15}; do curl localhost:8080/work -d name=job$i -d delay=$(expr $i % 9 + 1)s; done
Raw
worker_original.go
// Original code with Dispatcher
package main
import (
_ "expvar"
"flag"
"fmt"
"net/http"
_ "net/http/pprof"
"os"
"time"
)
// Job holds the attributes needed to perform unit of work.
type Job struct {
Name string
Delay time.Duration
}
// NewWorker creates takes a numeric id and a channel w/ worker pool.
func NewWorker(id int, workerPool chan chan Job) Worker {
return Worker{
id: id,
jobQueue: make(chan Job),
workerPool: workerPool,
quitChan: make(chan bool),
}
}
type Worker struct {
id int
jobQueue chan Job
workerPool chan chan Job
quitChan chan bool
}
func (w Worker) start() {
go func() {
for {
// Add my jobQueue to the worker pool.
w.workerPool <- w.jobQueue
select {
case job := <-w.jobQueue:
// Dispatcher has added a job to my jobQueue.
fmt.Printf("worker%d: started %s, blocking for %f seconds\n", w.id, job.Name, job.Delay.Seconds())
time.Sleep(job.Delay)
fmt.Printf("worker%d: completed %s!\n", w.id, job.Name)
case <-w.quitChan:
// We have been asked to stop.
fmt.Printf("worker%d stopping\n", w.id)
return
}
}
}()
}
func (w Worker) stop() {
go func() {
w.quitChan <- true
}()
}
// NewDispatcher creates, and returns a new Dispatcher object.
func NewDispatcher(jobQueue chan Job, maxWorkers int) *Dispatcher {
workerPool := make(chan chan Job, maxWorkers)
return &Dispatcher{
jobQueue: jobQueue,
maxWorkers: maxWorkers,
workerPool: workerPool,
}
}
type Dispatcher struct {
workerPool chan chan Job
maxWorkers int
jobQueue chan Job
}
func (d *Dispatcher) run() {
for i := 0; i < d.maxWorkers; i++ {
worker := NewWorker(i+1, d.workerPool)
worker.start()
}
go d.dispatch()
}
func (d *Dispatcher) dispatch() {
for {
select {
case job := <-d.jobQueue:
go func() {
fmt.Printf("fetching workerJobQueue for: %s\n", job.Name)
workerJobQueue := <-d.workerPool
fmt.Printf("adding %s to workerJobQueue\n", job.Name)
workerJobQueue <- job
}()
}
}
}
func requestHandler(w http.ResponseWriter, r *http.Request, jobQueue chan Job) {
// Make sure we can only be called with an HTTP POST request.
if r.Method != "POST" {
w.Header().Set("Allow", "POST")
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
// Parse the delay.
delay, err := time.ParseDuration(r.FormValue("delay"))
if err != nil {
http.Error(w, "Bad delay value: "+err.Error(), http.StatusBadRequest)
return
}
// Validate delay is in range 1 to 10 seconds.
if delay.Seconds() < 1 || delay.Seconds() > 10 {
http.Error(w, "The delay must be between 1 and 10 seconds, inclusively.", http.StatusBadRequest)
return
}
// Set name and validate value.
name := r.FormValue("name")
if name == "" {
http.Error(w, "You must specify a name.", http.StatusBadRequest)
return
}
// Create Job and push the work onto the jobQueue.
job := Job{Name: name, Delay: delay}
jobQueue <- job
// Render success.
w.WriteHeader(http.StatusCreated)
}
func main() {
var (
maxWorkers = flag.Int("max_workers", 5, "The number of workers to start")
maxQueueSize = flag.Int("max_queue_size", 100, "The size of job queue")
port = flag.String("port", "8080", "The server port")
)
flag.Parse()
// Create the job queue.
jobQueue := make(chan Job, *maxQueueSize)
// Start the dispatcher.
dispatcher := NewDispatcher(jobQueue, *maxWorkers)
dispatcher.run()
// Start the HTTP handler.
http.HandleFunc("/work", func(w http.ResponseWriter, r *http.Request) {
requestHandler(w, r, jobQueue)
})
log.Fatal(http.ListenAndServe(":"+*port, nil))
}
Raw
worker_refactored.go
package main
import (
_ "expvar"
"flag"
"fmt"
"log"
"net/http"
_ "net/http/pprof"
"time"
)
type job struct {
name string
duration time.Duration
}
type worker struct {
id int
}
func (w worker) process(j job) {
fmt.Printf("worker%d: started %s, working for %f seconds\n", w.id, j.name, j.duration.Seconds())
time.Sleep(j.duration)
fmt.Printf("worker%d: completed %s!\n", w.id, j.name)
}
func requestHandler(jobCh chan job, w http.ResponseWriter, r *http.Request) {
// Make sure we can only be called with an HTTP POST request.
if r.Method != "POST" {
w.Header().Set("Allow", "POST")
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
// Parse the durations.
duration, err := time.ParseDuration(r.FormValue("delay"))
if err != nil {
http.Error(w, "Bad delay value: "+err.Error(), http.StatusBadRequest)
return
}
// Validate delay is in range 1 to 10 seconds.
if duration.Seconds() < 1 || duration.Seconds() > 10 {
http.Error(w, "The delay must be between 1 and 10 seconds, inclusively.", http.StatusBadRequest)
return
}
// Set name and validate value.
name := r.FormValue("name")
if name == "" {
http.Error(w, "You must specify a name.", http.StatusBadRequest)
return
}
// Create Job and push the work onto the jobCh.
job := job{name, duration}
go func() {
fmt.Printf("added: %s %s\n", job.name, job.duration)
jobCh <- job
}()
// Render success.
w.WriteHeader(http.StatusCreated)
return
}
func main() {
var (
maxQueueSize = flag.Int("max_queue_size", 100, "The size of job queue")
maxWorkers = flag.Int("max_workers", 5, "The number of workers to start")
port = flag.String("port", "8080", "The server port")
)
flag.Parse()
// create job channel
jobCh := make(chan job, *maxQueueSize)
// create workers
for i := 0; i < *maxWorkers; i++ {
w := worker{i}
go func(w worker) {
for j := range jobCh {
w.process(j)
}
}(w)
}
// handler for adding jobs
http.HandleFunc("/work", func(w http.ResponseWriter, r *http.Request) {
requestHandler(jobCh, w, r)
})
log.Fatal(http.ListenAndServe(":"+*port, nil))
}
Raw
worker_standalone.go
package main
import (
_ "expvar"
"fmt"
"math/rand"
"sync"
"time"
)
type job struct {
name string
duration time.Duration
}
type worker struct {
id int
}
func (w worker) process(j job) {
fmt.Printf("worker%d: started %s, working for %fs\n", w.id, j.name, j.duration.Seconds())
time.Sleep(j.duration)
fmt.Printf("worker%d: completed %s!\n", w.id, j.name)
}
func main() {
wg := &sync.WaitGroup{}
jobCh := make(chan job)
// start workers
for i := 0; i < 10; i++ {
wg.Add(1)
w := worker{i}
go func(w worker) {
for j := range jobCh {
w.process(j)
}
wg.Done()
}(w)
}
// add jobs to queue
for i := 0; i < 100; i++ {
name := fmt.Sprintf("job-%d", i)
duration := time.Duration(rand.Intn(1000)) * time.Millisecond
fmt.Printf("adding: %s %s\n", name, duration)
jobCh <- job{name, duration}
}
// close jobCh and wait for workers to complete
close(jobCh)
wg.Wait()
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment