我从 google io 2010 中获取了负载均衡器代码，并为 Balancer 添加了优先级队列和同步锁定的实现。我故意设置 workFn函数延迟大于 requester这样我就可以看到待定值(value)将如何增加。我在 cli 中运行它并注意到在所有工作人员启动后，程序停止并为所有工作人员提供未决值 1 并且什么都不显示。有时我无法弄清楚错误在哪里completed只调用一次或两次。看起来像 <-b.done在选择案例中没有得到妥善处理。

package main

import (
    "container/heap"
    "fmt"
    "math/rand"
    "os"
    "sync"
    "time"
)

var nWorker int32 = 6

func main() {
    rchanel := make(chan Request)
    workers := Pool{
        {make(chan Request), 0, 0},
        {make(chan Request), 0, 1},
        {make(chan Request), 0, 2},
        {make(chan Request), 0, 3},
        {make(chan Request), 0, 4},
        {make(chan Request), 0, 5},
    }
    doneChan := make(chan *Worker)
    balancer := Balancer{workers, sync.Mutex{}, doneChan}
    for _, elem := range workers {
        go elem.work(doneChan)
    }
    go balancer.balance(rchanel)
    go requester(rchanel)

    var input string
    fmt.Scanln(&input)
}

type Request struct {
    fn func() int
    c  chan int
}

func requester(work chan Request) {
    c := make(chan int)
    for {
        time.Sleep(time.Duration(rand.Int31n(nWorker)) * 2e4)
        work <- Request{workFn, c}
        go func() {
            result := <-c
            fmt.Fprintf(os.Stderr, "Done: %v \n", result)
        }()
    }
}

func workFn() int {
    val := rand.Int31n(nWorker)
    time.Sleep(time.Duration(val) * 2e8)
    return int(val)
}

type Worker struct {
    requests chan Request
    pending  int
    index    int
}

func (w *Worker) work(done chan *Worker) {
    for {
        req := <-w.requests
        req.c <- req.fn()
        done <- w
    }
}

type Pool []*Worker

func (p Pool) Less(i, j int) bool {
    return p[i].pending < p[j].pending
}
func (p Pool) Swap(i, j int) {
    p[i], p[j] = p[j], p[i]
    p[i].index = i
    p[j].index = j
}
func (p Pool) Len() int { return len(p) }
func (p *Pool) Push(x interface{}) {
    n := len(*p)
    worker := x.(*Worker)
    worker.index = n
    *p = append(*p, worker)
}
func (p *Pool) Pop() interface{} {
    old := *p
    n := len(old)
    item := old[n-1]
    item.index = -1
    *p = old[0 : n-1]
    return item
}

type Balancer struct {
    pool Pool
    mu   sync.Mutex
    done chan *Worker
}

func (b *Balancer) dispatch(req Request) {
    b.mu.Lock()
    w := heap.Pop(&b.pool).(*Worker)
    w.requests <- req
    w.pending++
    heap.Push(&b.pool, w)
    b.mu.Unlock()
}
func (b *Balancer) completed(w *Worker) {
    b.mu.Lock()
    w.pending--
    heap.Remove(&b.pool, w.index)
    heap.Push(&b.pool, w)
    b.mu.Unlock()
}

func (b *Balancer) balance(work chan Request) {
    for {
        select {
        case req := <-work:
            b.dispatch(req)
            b.printStatus()
        case w := <-b.done:
            b.completed(w)
            b.printStatus()
        }
    }
}

func (b *Balancer) printStatus() {
    fmt.Fprintf(os.Stderr, "Status: %v %v %v %v %v %v\n", b.pool[0].pending, b.pool[1].pending, b.pool[2].pending, b.pool[3].pending, b.pool[4].pending, b.pool[5].pending)
}

最佳答案

问题是 balance() goroutine 最终在 dispatch() 中被阻塞在 w.requests <- req同时具体Worker在 work() 阻塞在 done <- w ，为 goroutine 运行产生死锁 balance() .

这是您需要的修复程序。 balance()需要在内部使用 goroutines。这将解决问题，因为现在例程是否阻塞在dispatch() 中并不重要。或 completed() , balance() 的主程序会继续select来自 channel

注意:这在 Playground 上不起作用，因为它会永远持续下去。

func (b *Balancer) balance(work chan Request) {
    for {
        select {
        case req := <-work:
            go func() {
                b.dispatch(req)
                b.printStatus()
            }()
        case w := <-b.done:
            go func() {
                b.completed(w)
                b.printStatus()
            }()
        }
    }
}

现在printStatus调用可以并发进行，需要利用mutex也一样，否则你会得到随机的panic

func (b *Balancer) printStatus() {
    b.mu.Lock()
    fmt.Fprintf(os.Stderr, "Status: %v %v %v %v %v %v\n", b.pool[0].pending, b.pool[1].pending, b.pool[2].pending, b.pool[3].pending, b.pool[4].pending, b.pool[5].pending)
    b.mu.Unlock()
}

现在如果我能弄清楚为什么 pending值(value)不断增加……据我所知，Worker.work()应该只允许 pending成为0或 1因为Worker必须等待 done <- w在它可以得到另一个 Request 之前来自 dispatch() .我相信这是期望的结果，不是吗？

关于go - 简单的负载均衡器无法正常工作，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/45881894/