job/queue.go
func (q *queue) Process() error {
	logger := q.logger.Add("method", "Process")
	q.mtx.Lock()

	if q.exit != nil {
		q.mtx.Unlock()
		return ErrAlreadyProcessing
	}

	num := int(q.conf.Workers)
	if num == 0 {
		num = runtime.NumCPU()
	}
	logger = logger.Add("workerNumber", num)

	// Make sure all workers can signal about errors simultaneously.
	q.exit = make(chan struct{}, num)
	q.exited = make(chan struct{}, 1)

	q.mtx.Unlock()

	q.workers = nil
	for i := 0; i < num; i++ {
		w := workerIO{
			job:    make(chan string, q.conf.WorkerBufLen),
			result: make(chan error, 1),
		}
		q.workers = append(q.workers, w)
		go q.processWorker(w)
	}

	err := q.processMain()

	// Stop the worker routines.

	for _, w := range q.workers {
		close(w.job)
	}

	for _, w := range q.workers {
		werr := <-w.result
		if werr != nil && err == ErrQueueClosed {
			err = werr
		}
	}

	q.exited <- struct{}{}

	q.mtx.Lock()
	q.exit = nil
	q.mtx.Unlock()

	return err
}
func (q *queue) Process() error {
	logger := q.logger.Add("method", "Process")
	q.mtx.Lock()

	if q.exit != nil {
		q.mtx.Unlock()
		return ErrAlreadyProcessing
	}

	num := int(q.conf.Workers)
	if num == 0 {
		num = runtime.NumCPU()
	}
	logger = logger.Add("workerNumber", num)

	// Make sure all workers can signal about errors simultaneously.
	q.exit = make(chan struct{}, num)
	q.exited = make(chan struct{}, 1)

	q.mtx.Unlock()

	q.workers = nil
	for i := 0; i < num; i++ {
		w := workerIO{
			job:    make(chan string, q.conf.WorkerBufLen),
			result: make(chan error, 1),
		}
		q.workers = append(q.workers, w)
		go q.processWorker(w)
	}

	err := q.processMain()

	// Stop the worker routines.

	for _, w := range q.workers {
		close(w.job)
	}

	for _, w := range q.workers {
		werr := <-w.result
		if werr != nil && err == ErrQueueClosed {
			err = werr
		}
	}

	q.exited <- struct{}{}

	q.mtx.Lock()
	q.exit = nil
	q.mtx.Unlock()

	return err
}
func ( q * queue ) Process ( ) error { logger := q . logger . Add ( " " , " " ) q . mtx . Lock ( ) if q . exit != nil { q . mtx . Unlock ( ) return ErrAlreadyProcessing } num := int ( q . conf . Workers ) if num == 0 { num = runtime . NumCPU ( ) } logger = logger . Add ( " " , num ) // Make sure all workers can signal about errors simultaneously. q . exit = make ( chan struct { } , num ) q . exited = make ( chan struct { } , 1 ) q . mtx . Unlock ( ) q . workers = nil for i := 0 ; i < num ; i ++ { w := workerIO { job : make ( chan string , q . conf . WorkerBufLen ) , result : make ( chan error , 1 ) , } q . workers = append ( q . workers , w ) go q . processWorker ( w ) } err := q . processMain ( ) // Stop the worker routines. for _ , w := range q . workers { close ( w . job ) } for _ , w := range q . workers { werr := <- w . result if werr != nil && err == ErrQueueClosed { err = werr } } q . exited <- struct { } { } q . mtx . Lock ( ) q . exit = nil q . mtx . Unlock ( ) return err }
// Process fetches active jobs and processes them in parallel. This function
// does not return until an error occurs or Close() is called.
Process fetches active jobs and processes them in parallel . This function does not return until an error occurs or Close () is called .