按顺序并发执行作业

Question

"What?" you ask, "That title doesn't make any sense."

Consider the following: Jobs with different ids may be processed asynchronously but jobs with the same id should be processed synchronously and in order from the queue.

My current implementation creates a go routine to handle the jobs for each specific id and looks something like this:

func FanOut() chan<- *Job {
    channel := make(chan *Job)
    routines = make(map[string]chan<- *Job)
    go func() {
        for j := range channel {
            r, found := routines[j.id]
            if !found {
                r = Routine()
                routines[j.id] = r
            }
            r <- j
        }
    }()
    return channel
}

This appears to work well (in current testing) but the creation of thousands of go routines might not be the best approach? Additionally the fan out code blocks unless a buffered channel is used.

Rather than a collection of go routines (above) I'm considering using a collection of sync.Mutex. The idea would be to have a pool of go routines which must first establish a lock on the mutex corresponding to the job id.

Are there any existing Go patterns suited to handling these requirements?

Is there a better approach?

Answer 1

Create a channel for each ID - perhaps a slice of channels or a map (indexed by ID). Each channel would have a go-routine that processes the jobs for that ID in order. Simple.

I wouldn't worry about creating too many go-routines. And I wouldn't use mutex - without getting into too much detail using channels and go-routines allows each job to only be processed by one go-routine at a time and avoids possibility of data races.

BTW I only added this as an answer as I am not permitted to add comments (yet?).