main.go

package main

import (
	"context"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"runtime"
	"sync"
)

type (
	job struct {
		proc func(context.Context)
		ctx  context.Context
	}

	Dispatcher struct {
		queue chan *job
		wg    sync.WaitGroup

		ctx    context.Context
		cancel context.CancelFunc
	}
)

const (
	maxWorkers = 3
	maxQueues  = 10000
)

func NewDispatcher() *Dispatcher {
	ctx, cancel := context.WithCancel(context.Background())

	d := &Dispatcher{
		queue:  make(chan *job, maxQueues),
		ctx:    ctx,
		cancel: cancel,
	}
	return d
}

func (d *Dispatcher) Add(proc func(context.Context)) {
	d.queue <- &job{proc: proc, ctx: d.ctx}
}

func (d *Dispatcher) AddWithContext(proc func(context.Context), ctx context.Context) {
	d.queue <- &job{proc: proc, ctx: ctx}
}

func (d *Dispatcher) Context() context.Context {
	return d.ctx
}

func (d *Dispatcher) Start() {
	d.wg.Add(maxWorkers)
	for i := 0; i < maxWorkers; i++ {
		go func() {
			defer d.wg.Done()
			for j := range d.queue {
				j.proc(j.ctx)
			}
		}()
	}
}

func (d *Dispatcher) Stop() {
	close(d.queue)
	d.wg.Wait()
	d.cancel()
}

func (d *Dispatcher) StopImmediately() {
	d.cancel()
	close(d.queue)
	d.wg.Wait()
}

func get(ctx context.Context) {
	url := ctx.Value("url").(string)
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		log.Fatal(err)
	}

	resp, err := http.DefaultClient.Do(req.WithContext(ctx))
	if err != nil {
		log.Print(err)
		return
	}
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Fatal(err)
	}

	log.Printf("Goroutine:%d, URL:%s (%d bytes)", runtime.NumGoroutine(), url, len(body))
}

func main() {
	d := NewDispatcher()

	d.Start()
	for i := 0; i < 100; i++ {
		url := fmt.Sprintf("http://placehold.it/%dx%d", i, i)
		ctx := context.WithValue(d.Context(), "url", url)
		d.AddWithContext(get, ctx)
	}
	d.Stop()
}