func (c Coordinator) Start() {
	for i := 0; i < c.numConsumer; i++ {
		go c.worker.start()
	}
}
Start workers to consume elements from queue.
package prosumer
import (
	"context"
	"log"
	"sync"
	"time"
)
type worker struct {
	*queue
	close     chan struct{}
	waitClose *sync.WaitGroup
	consumer      Consumer
	batchSize     int
	batchInterval time.Duration
	cb            Callback
}
func (w worker) start() {
	received := make([]Element, 0, w.batchSize)
	watchdog := time.NewTimer(w.batchInterval)
	doWork := func() {
		watchdog.Stop()
		if len(received) > 0 {
			err := w.consumer(received)
			w.cb(received, err)
			received = received[:0]
		}
		watchdog.Reset(w.batchInterval)
	}
loop:
	for {
		if e, ok := w.queue.dequeue(); ok {
			received = append(received, e)
			if len(received) >= w.batchSize {
				if enable_race {
					log.Println("doWork for reach batchSize")
				}
				doWork()
			} else {
				select {
				case <-watchdog.C:
					if enable_race {
						log.Println("doWork for reach batchInterval while dequeue")
					}
					doWork()
				default:
				}
			}
		} else {
			select {
			case <-w.close:
				doWork()
				watchdog.Stop()
				w.waitClose.Done()
				break loop
			case <-watchdog.C:
				if enable_race {
					log.Println("doWork for reach batchInterval")
				}
				doWork()
			}
		}
	}
}
type Coordinator struct {
	*queue
	*worker
	numConsumer int
	close     chan struct{}  
	waitClose *sync.WaitGroup
}
func NewCoordinator(config Config) Coordinator {
	var waitClose sync.WaitGroup
	waitClose.Add(config.numConsumer + 1)  
	closeCh := make(chan struct{})
	q := newQueue(config.bufferSize, config.rp, &waitClose)
	c := Coordinator{
		queue: q,
		worker: &worker{
			queue:         q,
			close:         closeCh,
			waitClose:     &waitClose,
			consumer:      config.consumer,
			batchSize:     config.batchSize,
			batchInterval: config.batchInterval,
			cb:            config.cb,
		},
		numConsumer: config.numConsumer,
		close:       closeCh,
		waitClose:   &waitClose,
	}
	return c
}