如何停止一个正在监听 RethinkDB changefeeds 的 goroutine？答案

【问题标题】：How to stop a goroutine that is listening for RethinkDB changefeeds?如何停止一个正在监听 RethinkDB changefeeds 的 goroutine？
【发布时间】：2016-11-05 15:44:57
【问题描述】：

我正在尝试弄清楚如何将 RethinkDB changefeeds 与 golang 一起使用。我的具体问题是如何停止一个监听更改的 goroutine 数据库。例如，参见下面的函数getData()。我从处理程序运行它通过调用go getData(c) 来运行。每当数据库更新时，记录是传递给通道c，然后传递给处理函数并发送给客户端使用 SSE 技术。我的问题是：当客户端断开连接时，我知道如何停止和退出处理函数；然而，运行 getData() 函数的 goroutine 继续运行。什么可以我要关闭它吗？根据stackoverflow上的其他答案，我能想到的一种解决方案是发送信号以关闭另一个通道上的 goroutine 并使用 select 语句来处理这个信号。例如，我可以替换

    for cur.Next(&rec) {
        c <- rec
    }

在下面的函数定义中：

    for cur.Next(&rec) {
        select {
         case <- closesignal:
            return
         default:
            c <- rec
        }
    }

其中，closesignal 是另一个通道，作为getData() 的第三个参数给出，并且当客户端断开连接时，处理程序会在此通道上发送一条消息。

这种方式的问题是：如果具体rethinkdb查询的结果怎么办从不更新。在这种情况下，不会进入for cur.Next(&rec) 循环，也不会使用closesignal。这个 goroutine 会继续运行吗？如果是这样，我该如何停止这个 goroutine？

getData() 函数

func getData(session *r.Session, c chan interface{}) {
    var rec interface{}

    changesOpts := r.ChangesOpts{
        IncludeInitial: true,
    }

    cur, err := r.DB(DBNAME).Table("test").Changes(changesOpts).Run(session)
    if err != nil {
        log.Println(err)
        return
    }
    defer cur.Close()

    defer func() {
        fmt.Println("exiting getData goroutine()...")
    }()


    for cur.Next(&rec) {
        c <- rec
    }

}

【问题讨论】：

标签： go rethinkdb goroutine

【解决方案1】：

您可以通过关闭光标来停止正在监听 changefeed 的 goroutine。例如，此代码将在关闭前监听 changefeed 10 秒：

go func() {
    time.Sleep(10 * time.Second)
    cur.Close()
}()

for cur.Next(&rec) {
    c <- rec
}

// Loop exits as the cursor has been closed

【讨论】：

谢谢！是的，这有帮助。你是写 gorethink 的同一个 Daniel Cannon。如果是这样，非常感谢。这是一个很棒的图书馆。
太好了，很高兴它现在可以工作了 :)（是的，我确实写了 GoRethink，谢谢！如果您有任何其他问题，请随时与 RethinkDB slack 联系）

【解决方案2】：

您应该通过context 停止changefeed。您可以将您的上下文传递给RunOpts。这将立即关闭您的更改提要。

【讨论】：

【解决方案3】：

我认为处理这种情况的最好方法是使用contexts 来关闭游标，并带有一个goroutine。这里有一些关于上下文的good documentation from the go blog。

这是一个带有上下文的实现示例：

func AreThereChanges(ctx context.Context) error {
    cursor, err := r.Table("something").
        Changes().
        Run(db.Session)

    if err != nil {
      //manage error
    }

    go func(ctx context.Context) {
        <-ctx.Done()
        cursor.Close()
    }(ctx)

    var changeFeed map[string]interface{}
    for cursor.Next(&changeFeed) {
        if changeFeed["old_val"] != nil {
          //do something
        }

        if changeFeed["new_val"] != nil {
          //do something
        }
    }

【讨论】：