我正在尝试学习如何使用 channel 在 Go 中为我的其他项目之一创建队列。我的其他项目基本上是对数据库行进行排队，然后使用行中的详细信息对数据库进行数字运算。

我不希望同一行同时在一个工作程序中处理，因此它需要检查一个工作程序当前是否正在处理该特定行 ID，如果是，则等待它完成。如果不是相同的行ID，它可以异步运行，但我也想限制可以同时运行的异步 worker 的数量。在我下面的代码中，我目前试图将其限制为三名 worker 。

这是我所拥有的:

package main

import (
    "log"
    "strconv"
    "time"
)

// RowInfo holds the job info
type RowInfo struct {
    id int
}

// WorkerCount holds how many workers are currently running
var WorkerCount int

// WorkerLocked specifies whether a row ID is currently processing by a worker
var WorkerLocked map[string]bool

// Process the RowInfo
func worker(row RowInfo) {
    rowID := strconv.Itoa(row.id)

    WorkerCount++
    WorkerLocked[rowID] = true

    time.Sleep(1 * time.Second)
    log.Printf("ID rcvd: %d", row.id)

    WorkerLocked[rowID] = false
    WorkerCount--
}

// waiter will check if the row is already processing in a worker
// Block until it finishes completion, then dispatch
func waiter(row RowInfo) {
    rowID := strconv.Itoa(row.id)
    for WorkerLocked[rowID] == true {
        time.Sleep(1 * time.Second)
    }

    go worker(row)
}

func main() {
    jobsQueue := make(chan RowInfo, 10)
    WorkerLocked = make(map[string]bool)

    // Dispatcher waits for jobs on the channel and dispatches to waiter
    go func() {
        // Wait for a job
        for {
            // Only have a max of 3 workers running asynch at a time
            for WorkerCount > 3 {
                time.Sleep(1 * time.Second)
            }

            job := <-jobsQueue
            go waiter(job)
        }
    }()

    // Test the queue, send some data
    for i := 0; i < 12; i++ {
        r := RowInfo{
            id: i,
        }
        jobsQueue <- r
    }

    // Prevent exit!
    for {
        time.Sleep(1 * time.Second)
    }
}

我收到此错误，但这是一个间歇性问题，因为有时当我运行它时它似乎可以正常工作。是否存在竞争条件？:

go run main.go
panic: runtime error: invalid memory address or nil pointer dereference
[signal 0xb code=0x1 addr=0x8 pc=0x4565e7]

goroutine 37 [running]:
main.worker(0x5)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:25 +0x94
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 1 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.main()
    /home/piiz/go/src/github.com/zzz/asynch/main.go:73 +0xf8

goroutine 5 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.main.func1(0xc82008c000)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:55 +0x2d
created by main.main
    /home/piiz/go/src/github.com/zzz/asynch/main.go:61 +0xa0

goroutine 35 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0x2)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 36 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0x4)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 34 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0x1)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 12 [runnable]:
runtime.goexit1()
    /usr/local/go/src/runtime/proc1.go:1732
runtime.goexit()
    /usr/local/go/src/runtime/asm_amd64.s:1697 +0x6
created by main.main.func1
    /home/piiz/go/src/github.com/zzz/asynch/main.go:59 +0x8c

goroutine 19 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0x8)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 20 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0x0)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 16 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0x9)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 33 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0x3)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 18 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0x7)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 22 [sleep]:
time.Sleep(0x3b9aca00)
    /usr/local/go/src/runtime/time.go:59 +0xf9
main.worker(0xa)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:27 +0xa1
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb

goroutine 49 [runnable]:
main.worker(0x6)
    /home/piiz/go/src/github.com/zzz/asynch/main.go:21
created by main.waiter
    /home/piiz/go/src/github.com/zzz/asynch/main.go:42 +0xbb
exit status 2

无论如何，我仍在学习，所以如果您查看我的代码并“搞什么鬼”，好吧，我不会感到惊讶 :) 也许我完全错误地解决了这个问题。谢谢。

最佳答案

如果您要使用 WorkerLocked map ，您需要使用 sync 包来保护对它的访问。您还需要以相同的方式(或使用原子操作)保护 WorkerCount。做类似的事情也会使 sleep 变得不必要(使用条件变量)。

更好的是，有 3 个(或许多)工作人员等待行使用 channel 进行工作。然后，您可以将行分发给不同的工作人员，这样特定的行始终由特定的工作人员处理(例如，使用 row.id % 3 来确定将行发送到哪个工作人员/ channel )。

关于Go:学习 channel &排队， fatal error ，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/35329276/