goroutine_ack.go

package main

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

var workerNum = 2
var msgNum = 10

type Job struct {
	i   int
	msg string
}

func processJob(worker int, job Job) {
	rand.Seed(time.Now().UnixNano())
	time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
	fmt.Println("worker", worker, job.i, job.msg)
}

func processAck(worker int, jobs <-chan Job, done chan<- bool) {
	for {
		job, more := <-jobs

		if more {
			processJob(worker, job)
		} else {
			fmt.Println("worker", worker, "complete")
			done <- true
			return
		}
	}
}

func processNoAck(worker int, jobs <-chan Job) {
	for job := range jobs {
		processJob(worker, job)
	}
}

// this worker pool ignores the finished state and doesn't wait for the messages to be deleted
func runNoAck() {
	jobs := make(chan Job)

	for i := 0; i < workerNum; i++ {
		go processNoAck(i, jobs)
	}

	for i := 0; i < msgNum; i++ {
		fmt.Println("start no ack", i)
		jobs <- Job{i: i, msg: "no ack"}
	}
	close(jobs)

	fmt.Println("Finished No ACK")
}

func runAck() {
	jobs := make(chan Job)
	done := make(chan bool)

	for i := 0; i < workerNum; i++ {
		go processAck(i, jobs, done)
	}

	for i := 0; i < msgNum; i++ {
		fmt.Println("start ack", i)
		jobs <- Job{i: i, msg: "ack"}
	}
	close(jobs)

	for i := 0; i < workerNum; i++ {
		<-done
	}
	close(done)
	fmt.Println("Finished Ack")
}

func main() {
	runNoAck()
	runAck()
}