基于生產(chǎn)者-消費(fèi)者的設(shè)計(jì)。

任務(wù)數(shù)據(jù)結(jié)構(gòu) Task 自帶方法 exec() 負(fù)責(zé)執(zhí)行Task任務(wù)的任意函數(shù) f()，Pool的run()開(kāi)啟N個(gè)消費(fèi)者協(xié)程監(jiān)聽(tīng)共享無(wú)緩沖channel，生產(chǎn)者協(xié)程封裝數(shù)據(jù)id和函數(shù)t到數(shù)據(jù)結(jié)構(gòu)Task，導(dǎo)入channel執(zhí)行。

package main

import (
    "fmt"
    "time"
    "math/rand"
)

type Task struct {
    ProducerId int
    TaskId int
    f func() time.Time
}

func NewTask(id int, taskid int,f func() time.Time) *Task {
    return &Task{
        ProducerId: id,
        TaskId: taskid,
        f: f,
    }
}

type Pool struct {
    workerNum int
    workerChan chan *Task
}

func NewPool(num int) *Pool {
    return &Pool{
        workerNum: num,
        workerChan: make(chan *Task),
    }
}

func (p *Pool) worker(id int) {
    for task := range p.workerChan {
        fmt.Println(task.f(),": ConsumerId:", id, "ProducerId:", task.ProducerId, "TaskId:", task.TaskId,"is done")
    }
}

func (p *Pool) Run() {
    for i := 0; i < p.workerNum; i++ {
        go p.worker(i)
    }
}

func task() time.Time {
    return time.Now()
}

func main() {
    p := NewPool(10)
    p.Run()
    for producerId :=0; producerId < 15; producerId++ {
        go func(producerId int){
            for i := 0; i < 50; i++ {
                p.workerChan<-NewTask(producerId,rand.Int(),task)
            }
        }(producerId)
    }
    time.Sleep(3*time.Second)
}