Golang 非阻塞通道不起作用

Question

我第一次在 go 中使用 goroutines 和 channels 并且不再进一步。

我有一个 websocket 连接，每次用户连接时都会产生一个新的 goroutine。现在我想在用户断开 websocket 连接时停止这个 goroutine。

为了管理停止信号，我创建了一张通道图。每个条目都可以通过用户的 websocket 连接来识别。我将 websocket 连接、停止信号的通道映射和其他两个参数传递给 goroutine。但是 goroutine 没有从退出通道接收任何值，我不知道为什么。

这里是main包的相关代码：

package main

import (
    "net/http"
    "time"

    "github.com/gorilla/websocket"
)

func wsHandler(w http.ResponseWriter, r *http.Request) {
    ...
    ws, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        return
    }
    defer ws.Close()

    data.Quit[ws] = make(chan bool)

    data.DB.ListenToTable(data.GetTableName(source), channel, data.Quit, ws)

    for {
        if _, _, err := ws.NextReader(); err != nil {
            data.Quit[ws] <- true

            ws.Close()
            break
        }
    }
}

以及创建goroutine的data包的代码：

package data

var Quit = make(map[*websocket.Conn](chan bool))

func (db *rethinkDB) ListenToTable(name string, ch chan Data, quit map[*websocket.Conn](chan bool), ws *websocket.Conn) {
    go func(name string, ws *websocket.Conn) {
        for {
            select {
            case <-quit[ws]:
                fmt.Println("QUIT GOROUTINE")
                break
            default:
                res, err := r.Table(name).Changes().Run(db.session)
                if err != nil {
                    log.Fatalln(err)
                }

                var response DataFeed
                for res.Next(&response) {
                    response.NewVal.WS = ws
                    ch <- response.NewVal
                }

                if res.Err() != nil {
                    log.Println(res.Err())
                }
            }
        }
    }(name, ws)
}

我也尝试过缓冲通道或将通道而不是通道映射传递给 goroutine，但没有成功。 fmt.Println("QUIT GOROUTINE") 命令永远不会被调用，goroutine 也不会被停止。

我希望有人可以帮助我，如果这个问题已经被问过，但我还没有找到解决我的问题的解决方案，我很抱歉。

Answer 1

首先让事情变得更容易：

据我所知，您不需要退出频道的全局寄存器。只需在 main 中创建一个ch := make(chan bool)，将其传递给ListenToTable（而不是整个频道图）并在选择中使用它。如果您想退出，请在主 close(ch) 中。但正如你所说，这并不能解决你的问题。

理论上，关闭 go 例程是正确的。我获取了您的示例代码并从中制作了以下可运行代码：

package main

import (
    "fmt"
    "time"
)

func main() {

    chClose := make(chan bool)
    channel := make(chan string)

    ListenToTable("somestring", channel, chClose)

    time.Sleep(3 * time.Second)
    chClose <- true
    time.Sleep(1 * time.Second)
}

func ListenToTable(name string, ch chan string, chClose chan bool) {
    go func(name string) {
        for {
            select {
            case <-chClose:
                fmt.Println("QUIT GOROUTINE")
                return  // VERY IMPORTANT: not break!
            default:
            }
        }
    }(name)
}

问题一定出在代码中的其他内容上，可能被default 部分中的某些内容阻塞，甚至没有执行选择。尝试在select { 之前打印fmt.Println("something")。如果没有定期打印，那么您就有答案了。

还有一件事：正如上面代码中所评论的，您不能使用单个 break 来突破 for { select { ... } }。您需要使用 return （退出函数）或其他策略（如 Adrian 在 cmets 中建议的使用标签中断）。 break 只会退出select，不会退出for 循环。