基于Golang 高并發(fā)問題的解決方案

//==================================
//  * Name：Jerry
//  * Tel：18017448610
//  * DateTime：2019/2/24 14:02
//==================================
package main
import (
	"github.com/lunny/log"
	"runtime"
	"sync"
	"time"
)
//工廠模型
type Factory struct {
	Wg        *sync.WaitGroup //任務(wù)監(jiān)控系統(tǒng)
	MaxWorker int             //最大機(jī)器數(shù)
	MaxJobs   int             //最大工作數(shù)量
	JobQueue  chan int        //工作隊(duì)列管道
	Quit      chan bool       //是否關(guān)閉機(jī)器
}
//創(chuàng)建工廠模型
func NewFactory(maxWorker int, wg *sync.WaitGroup) Factory {
	return Factory{
		Wg:        wg,                        //引用任務(wù)監(jiān)控系統(tǒng)
		MaxWorker: maxWorker,                 //機(jī)器數(shù)量（數(shù)量多少，根據(jù)服務(wù)器性能而定）
		JobQueue:  make(chan int, maxWorker), //工作管道，數(shù)量大于等于機(jī)器數(shù)
		Quit:      make(chan bool),
	}
}
//設(shè)置最大訂單數(shù)量
func (f *Factory) SetMaxJobs(taskNum int) {
	f.MaxJobs = taskNum
}
//開始上班
func (f *Factory) Start() {
	//機(jī)器開機(jī)，MaxWorker
	for i := 0; i  f.MaxWorker; i++ {
		//每一臺機(jī)器開啟后，去工作吧
		go func() {
			//等待下發(fā)命令
			for {
				select {
				case i := -f.JobQueue:
					//接到工作，開工！
					f.doWork(i)
				case -f.Quit:
					log.Println("機(jī)器關(guān)機(jī)")
					return
				}
			}
		}()
	}
}
//分配每個(gè)任務(wù)到管道中
func (f *Factory) AddTask(taskNum int) {
	//系統(tǒng)監(jiān)控任務(wù) +1
	f.Wg.Add(1)
	//分配任務(wù)到管道中
	f.JobQueue - taskNum
}
//模擬耗時(shí)工作
func (f *Factory) doWork(taskNum int) {
	//生產(chǎn)產(chǎn)品的工作
	time.Sleep(200 * time.Millisecond)
	//完成工作報(bào)告
	f.Wg.Done()
	//log.Println("完工：", taskNum)
}
//創(chuàng)建工廠
func Begin() {
	//配置工作核數(shù)
	gomaxprocs := runtime.GOMAXPROCS(runtime.NumCPU())
	log.Println("核數(shù)：", gomaxprocs)
	//配置監(jiān)控系統(tǒng)
	wg := new(sync.WaitGroup)
	//開工廠
	factory := NewFactory(1000, wg)
	//訂單量
	factory.SetMaxJobs(10000)
	//開始上班
	factory.Start()
	log.Println("開始生產(chǎn)")
	//講所有的訂單，添加到任務(wù)隊(duì)列
	for i := 0; i  factory.MaxJobs; i++ {
		factory.AddTask(i)
	}
	factory.Wg.Wait()
	log.Println("所有訂單任務(wù)生產(chǎn)完成")
}

func Benchmark_Begin(b *testing.B) {
 Begin()
}

//任務(wù)的請求
type MtaskRequest struct {
    Ceshi int
    // [redacted]
}
 
//job隊(duì)列+work池
var (
    MaxWorker = os.Getenv("MAX_WORKERS")
    MaxQueue  = os.Getenv("MAX_QUEUE")
)
 
// Job represents the job to be run
type Job struct {
    MtaskRequest MtaskRequest
}
 
// A buffered channel that we can send work requests on.
 
// var JobQueue chan Job ---這樣申明會卡主，沒有初始化
var JobQueue = make(chan Job)
 
// Worker represents the worker that executes the job
type Worker struct {
    WorkerPool chan chan Job
    JobChannel chan Job
    quit       chan bool
}
 
func NewWorker(workerPool chan chan Job) Worker {
    return Worker{
        WorkerPool: workerPool,
        JobChannel: make(chan Job),
        quit:       make(chan bool)}
}
 
// Stop signals the worker to stop listening for work requests.
func (w Worker) Stop() {
    go func() {
        w.quit - true
    }()
}
 
type Dispatcher struct {
    // A pool of workers channels that are registered with the dispatcher
    WorkerPool chan chan Job
    maxWorkers int
}
 
func NewDispatcher(maxWorkers int) *Dispatcher {
    pool := make(chan chan Job, maxWorkers)
    return Dispatcher{WorkerPool: pool, maxWorkers: maxWorkers}
}
 
// Start method starts the run loop for the worker, listening for a quit channel in
// case we need to stop it
func (w Worker) Start() {
    go func() {
        for {
            // register the current worker into the worker queue.
            w.WorkerPool - w.JobChannel
            select {
            case -w.JobChannel:
                time.Sleep(5 * time.Second)
                // we have received a work request.
                fmt.Println("調(diào)起worker")
            case -w.quit:
                // we have received a signal to stop
                return
                //不能寫default
            }
        }
    }()
}
 
func (d *Dispatcher) Run() {
    //啟動一定數(shù)量的worker
    fmt.Println("啟動一定數(shù)量的worker")
    for i := 0; i  d.maxWorkers; i++ {
        worker := NewWorker(d.WorkerPool)
        worker.Start()
    }
 
    go d.dispatch()
}
 
//分派任務(wù)
func (d *Dispatcher) dispatch() {
    for {
        select {
        case job := -JobQueue: //接收一個(gè)job請求
            fmt.Println("JobQueue 收到請求")
 
            go func(job Job) {
                // try to obtain a worker job channel that is available.
                // this will block until a worker is idle
                jobChannel := -d.WorkerPool
                // dispatch the job to the worker job channel
                jobChannel - job
            }(job)
        }
    }
}
 
//接收到紅包數(shù)據(jù)
func (this *TaskRedbao) UserGetRedbao(red_id, uid, shop_id, rand_arr, Amoney string) error {
    fmt.Println("收到 接收到紅包數(shù)據(jù) http請求")
    mtaskRequest := MtaskRequest{67}
    work := Job{MtaskRequest: mtaskRequest}
 
    JobQueue - work
    return nil
}