我正在尝试从接收端实现优雅的 channel 关闭。

是的，我知道这违反了 channel 关闭规则:

...don't close a channel from the receiver side and don't close a channel if the channel has multiple concurrent senders.

但是我想实现这样的逻辑。不幸的是，我在很多情况下都没有陷入死锁问题:应用程序只是无限期地挂起，试图再次锁定相同的锁定 Mutex。

所以，我有 2 个协程:

• 将写入 channel 的一个
• 另一个将接收数据 + 将从接收端关闭 channel 。

我的 channel 用 sync.Mutex 和 closed bool 标志包裹在结构中:

type Chan struct {
    sync.Mutex // can be replaced with deadlock.Mutex from "github.com/sasha-s/go-deadlock"
    data           chan int
    closed         bool
}

此结构上的所有 Send()、Close()、IsClosed() 操作均由 Mutex 并防止重复锁定具有非线程安全方法版本(send()、close()、isClosed())。

完整源代码:

package main

import (
    "log"
    "net/http"
    "sync"
)

func main() {
    log.Println("Start")

    ch := New(0) // unbuffered channel to expose problem faster

    wg := sync.WaitGroup{}
    wg.Add(2)

    // send data:
    go func(ch *Chan) {
        for i := 0; i < 100; i++ {
            ch.Send(i)
        }
        wg.Done()
    }(ch)

    // receive data and close from receiver side:
    go func(ch *Chan) {
        for data := range ch.data {
            log.Printf("Received %d data", data)
            // Bad practice: I want to close the channel from receiver's side:
            if data > 50 {
                ch.Close()
                break
            }
        }
        wg.Done()
    }(ch)

    wg.Wait()
    log.Println("End")
}

type Chan struct {
    deadlock.Mutex //sync.Mutex
    data           chan int
    closed         bool
}

func New(size int) *Chan {
    defer func() {
        log.Printf("Channel was created")
    }()
    return &Chan{
        data: make(chan int, size),
    }
}

func (c *Chan) Send(data int) {
    c.Lock()
    c.send(data)
    c.Unlock()
}

func (c *Chan) Close() {
    c.Lock()
    c.close()
    c.Unlock()
}

func (c *Chan) IsClosed() bool {
    c.Lock()
    defer c.Unlock()
    return c.isClosed()
}

// send is internal non-threadsafe api.
func (c *Chan) send(data int) {
    if !c.closed {
        c.data <- data
        log.Printf("Data %d was sent", data)
    }
}

// close is internal non-threadsafe api.
func (c *Chan) close() {
    if !c.closed {
        close(c.data)
        c.closed = true
        log.Println("Channel was closed")
    } else {
        log.Println("Channel was already closed")
    }
}

// isClosed is internal non-threadsafe api.
func (c *Chan) isClosed() bool {
    return c.closed
}

您可以在 sandbox 中运行此程序.

在本地机器上，在少量运行中，30 秒后输出将是(使用 deadlock.Mutex 而不是 sync.Mutex):

2018/04/01 11:26:22 Data 50 was sent
2018/04/01 11:26:22 Received 50 data
2018/04/01 11:26:22 Data 51 was sent
2018/04/01 11:26:22 Received 51 data
POTENTIAL DEADLOCK:
Previous place where the lock was grabbed
goroutine 35 lock 0xc42015a040
close-from-receiver-side/closeFromReceiverSideIsBadPractice.go:71 main.(*Chan).Send { c.Lock() } <<<<<
close-from-receiver-side/closeFromReceiverSideIsBadPractice.go:30 main.main.func1 { ch.Send(i) }

Have been trying to lock it again for more than 30s
goroutine 36 lock 0xc42015a040
close-from-receiver-side/closeFromReceiverSideIsBadPractice.go:77 main.(*Chan).Close { c.Lock() } <<<<<
close-from-receiver-side/closeFromReceiverSideIsBadPractice.go:44 main.main.func2 { ch.Close() }

为什么会发生这种死锁以及如何修复此实现以避免死锁？

关闭发送方的 channel 不是解决办法。所以，这不是我的问题的解决方法:Example of closing channel from sender side .

最佳答案

Send 获取锁，然后尝试向 channel 发送数据。这可能发生在第 50 次接收操作之后。不会有更多的接收，所以 c.data <- data永远阻塞，因此 Mutex 永远持有。

要取消，请使用另一个 channel (而不是 bool 值)和 Send 中的 select 语句。您还可以利用 the context package .

关于go - 从接收端关闭 channel : deadlock when accessing sync. 来自多个 goroutine 的互斥量，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/49596347/