package main
import (
	"fmt"
	"net/http"
)
func main() {
	myHandler := MyHandler{}
	http.Handle("/", &myHandler)
	http.ListenAndServe(":8080", nil)
}
type MyHandler struct {
}
func (h *MyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	w.Write([]byte("Hello Go"))
}

type Job struct {
    DoneChan  chan struct{}
    handleJob func(j FlowJob) error //具體的處理邏輯
}

// 消費(fèi)者從隊(duì)列中取出該job時(shí) 執(zhí)行具體的處理邏輯
func (job *Job) Execute() error {
    fmt.Println("job start to execute ")
    return job.handleJob(job)
}
// 執(zhí)行完Execute后，調(diào)用該函數(shù)以通知主線程中等待的job
func (job *Job) Done() {
    job.DoneChan <- struct{}{}
    close(job.DoneChan)
}
// 工作單元等待自己被消費(fèi)
func (job *Job) WaitDone() {
    select {
    case <-job.DoneChan:
	return
    }
}

type JobQueue struct {
    mu         sync.Mutex
    noticeChan chan struct{}
    queue      *list.List
    size       int
    capacity   int
}

// 初始化隊(duì)列
func NewJobQueue(cap int) *JobQueue {
    return &JobQueue{
	capacity: cap,
	queue:    list.New(),
	noticeChan: make(chan struct{}, 1),
    }
}
// 工作單元入隊(duì)
func (q *JobQueue) PushJob(job *Job) {
    q.mu.Lock()
    defer q.mu.Unlock()
    q.size++
    if q.size > q.capacity {
	q.RemoveLeastJob()
    }
    q.queue.PushBack(job)
    q.noticeChan <- struct{}{}
}
// 工作單元出隊(duì)
func (q *JobQueue) PopJob() *Job {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.size == 0 {
		return nil
	}
	q.size--
	return q.queue.Remove(q.queue.Front()).(*Job)
}
// 移除隊(duì)列中的最后一個(gè)元素。
// 一般在容量滿時(shí)，有新job加入時(shí)，會(huì)移除等待最久的一個(gè)job
func (q *JobQueue) RemoveLeastJob() {
	if q.queue.Len() != 0 {
		back := q.queue.Back()
		abandonJob := back.Value.(*Job)
		abandonJob.Done()
		q.queue.Remove(back)
	}
}
// 消費(fèi)線程監(jiān)聽隊(duì)列的該通道，查看是否有新的job需要消費(fèi)
func (q *JobQueue) waitJob() <-chan struct{} {
    return q.noticeChan
}

type WorkerManager struct {
    jobQueue *JobQueue
}
func NewWorkerManager(jobQueue *JobQueue) *WorkerManager {
    return &WorkerManager{
	jobQueue: jobQueue,
    }
}
func (m *WorkerManager) createWorker() error {
    go func() {
	fmt.Println("start the worker success")
	var job FlowJob
	for {
            select {
                case <-m.jobQueue.waitJob():
		fmt.Println("get a job from job queue")
                job = m.jobQueue.PopJob()
		fmt.Println("start to execute job")
		job.Execute()
                fmt.Print("execute job done")
		job.Done()
            }
	}
    }()
    return nil
}

package main
import (
    "container/list"
    "fmt"
    "net/http"
    "sync"
)
func main() {
    flowControl := NewFlowControl()
    myHandler := MyHandler{
	flowControl: flowControl,
    }
    http.Handle("/", &myHandler)
    http.ListenAndServe(":8080", nil)
}
type MyHandler struct {
    flowControl *FlowControl
}
func (h *MyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	fmt.Println("recieve http request")
	job := &Job{
            DoneChan: make(chan struct{}, 1),
            handleJob: func(job *Job) error {
		w.Header().Set("Content-Type", "application/json")
		w.Write([]byte("Hello World"))
		return nil
            },
	}
	h.flowControl.CommitJob(job)
	fmt.Println("commit job to job queue success")
	job.WaitDone()
}
type FlowControl struct {
    jobQueue *JobQueue
    wm       *WorkerManager
}
func NewFlowControl() *FlowControl {
    jobQueue := NewJobQueue(10)
    fmt.Println("init job queue success")
    m := NewWorkerManager(jobQueue)
    m.createWorker()
    fmt.Println("init worker success")
    control := &FlowControl{
	jobQueue: jobQueue,
	wm:       m,
    }
    fmt.Println("init flowcontrol success")
    return control
}
func (c *FlowControl) CommitJob(job *Job) {
    c.jobQueue.PushJob(job)
    fmt.Println("commit job success")
}