we have a async task scheduler system, which use golang's exec.Command to execute php scripts. Each time, the scheduler fetch a task from message queue, it will create a new exec.Command to execute the task.
Sometimes, there are almost thousands of tasks need to be executed at once. In this case, the scheduler will create thousands of exec.Command, then destroy them after a while.
I wanna know if there is a way to create a process pool(or something like this) so that we can reuse the exec.Command to reduce the cost of create new child process.
ps: I noticed that exec.Command can't be reused after called run.
UPATE the current logic would like below:
func nsqMessageHandler(msg *nsq.Message) error{
var task Task
err:= json.Unmarshal(msg.Body,&task)
....
cmd:=exec.Command("php",task.Payload...)
err:=cmd.Start()
....
err= cmd.Run()
err=cmd.Process.Kill()
...
}
Definitely yes. You can create a blocking channel which you submit the jobs to. Then you can create a set of "workers" by spawning goroutines that take the job<-channel as input and have for example an output channel to publish progress or results. The workers now just read on the channel and once one of it gets a job it will be working.
You want to prevent the workers from shutting down right at the beginning, so you need some method of blocking them. One way to solve this problem is to use a waitgroup and let the workers decrease the waitgroup index by one if one of them is off. Furthermore you want to stop the workers but you cannot stop a goroutine from outside - so you have to implement a job which you can pass to a worker which forces them to stop by their own.
An example from gobyexample.com
// In this example we'll look at how to implement
// a _worker pool_ using goroutines and channels.
package main
import "fmt"
import "time"
// Here's the worker, of which we'll run several
// concurrent instances. These workers will receive
// work on the `jobs` channel and send the corresponding
// results on `results`. We'll sleep a second per job to
// simulate an expensive task.
func worker(id int, jobs <-chan int, results chan<- int) {
for j := range jobs {
fmt.Println("worker", id, "started job", j)
time.Sleep(time.Second)
fmt.Println("worker", id, "finished job", j)
results <- j * 2
}
}
func main() {
// In order to use our pool of workers we need to send
// them work and collect their results. We make 2
// channels for this.
jobs := make(chan int, 100)
results := make(chan int, 100)
// This starts up 3 workers, initially blocked
// because there are no jobs yet.
for w := 1; w <= 3; w++ {
go worker(w, jobs, results)
}
// Here we send 5 `jobs` and then `close` that
// channel to indicate that's all the work we have.
for j := 1; j <= 5; j++ {
jobs <- j
}
close(jobs)
// Finally we collect all the results of the work.
for a := 1; a <= 5; a++ {
<-results
}
}