Golang - 为到不同服务器的多个连接扩展 websocket 客户端

Question

我有一个 websocket 客户端。实际上，它比下面显示的基本代码复杂得多。我现在需要扩展此客户端代码以打开到多个服务器的连接。最终，从服务器接收到消息时需要执行的任务是相同的。处理这个问题的最佳方法是什么？正如我上面所说，接收消息时执行的实际代码比示例中显示的要复杂得多。

package main

import (
        "flag"
        "log"
        "net/url"
        "os"
        "os/signal"
        "time"

        "github.com/gorilla/websocket"
)

var addr = flag.String("addr", "localhost:1234", "http service address")

func main() {
        flag.Parse()
        log.SetFlags(0)

        interrupt := make(chan os.Signal, 1)
        signal.Notify(interrupt, os.Interrupt)

        // u := url.URL{Scheme: "ws", Host: *addr, Path: "/echo"}
        u := url.URL{Scheme: "ws", Host: *addr, Path: "/"}
        log.Printf("connecting to %s", u.String())

        c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
        if err != nil {
                log.Fatal("dial:", err)
        }
        defer c.Close()

        done := make(chan struct{})

        go func() {
                defer close(done)
                for {
                        _, message, err := c.ReadMessage()
                        if err != nil {
                                log.Println("read:", err)
                                return
                        }
                        log.Printf("recv: %s", message)
                }
        }()

        ticker := time.NewTicker(time.Second)
        defer ticker.Stop()

        for {
                select {
                case <-done:
                        return
                case t := <-ticker.C:
                        err := c.WriteMessage(websocket.TextMessage, []byte(t.String()))
                        if err != nil {
                                log.Println("write:", err)
                                return
                        }
                case <-interrupt:
                        log.Println("interrupt")

                        // Cleanly close the connection by sending a close message and then
                        // waiting (with timeout) for the server to close the connection.
                        err := c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
                        if err != nil {
                                log.Println("write close:", err)
                                return
                        }
                        select {
                        case <-done:
                        case <-time.After(time.Second):
                        }
                        return
                }
        }
}

Answer 1

与每个不同服务器的通信是否完全独立于其他服务器？如果是的话，我会以这样的方式四处走动：

在 main 中创建一个带有取消函数的 context
在 main 中创建一个 waitgroup 来跟踪启动的 goroutines
对于每个服务器，添加到等待组，从传递上下文和等待组引用的主函数启动一个新的 goroutine
main 进入 for/select 循环侦听信号，如果信号到达，调用 cancelfunc 并等待等待组。
main 也可以监听来自 goroutines 的结果 chan 并且可能自己打印结果 goroutines 不应该直接这样做。
每个 goroutine 都有我们所说的工作组、上下文和可能的 return 结果的引用。现在，如果 goroutine 必须只做一件事，或者它是否需要做一系列事情，这个方法就会分裂。对于第一种方法
如果只做一件事，我们会采用类似 here 中描述的方法（观察到异步他会依次启动一个新的 goroutine 来执行 DoSomething() 步骤，这将return 频道上的结果）这使它能够随时接受取消信号。由你来决定你想要多无阻塞以及你想要多快响应取消 signals.Also 将关联的上下文传递给 goroutines 的好处是你可以调用大多数库函数的上下文启用版本。例如，如果您希望您的拨号超时为 1 分钟，您可以创建一个新的上下文，该上下文从传递的超时开始，然后使用该超时创建 DialContext。这允许拨号从超时或父级（您在 main 中创建的）上下文的 cancelfunc 被调用时停止。
如果需要做更多的事情，我通常更喜欢用 goroutine 做一件事，让它调用一个新的 goroutine 来执行下一步（将所有引用传递到管道中）然后退出。

这种方法可以很好地适应取消，并能够在任何步骤停止流水线，并支持带有交易的上下文，很容易处理可能需要很长时间的步骤。

Answer 2

修改中断处理以在中断时关闭通道。这允许多个 goroutines 通过等待通道关闭来等待事件。

shutdown := make(chan struct{})
interrupt := make(chan os.Signal, 1)
signal.Notify(interrupt, os.Interrupt)
go func() {
    <-interrupt
    log.Println("interrupt")
    close(shutdown)
}()

将每个连接代码移动到一个函数中。这段代码是从问题中复制粘贴的，有两个变化：中断通道被关闭通道替换；该函数在函数完成时通知 sync.WaitGroup。

func connect(u string, shutdown chan struct{}, wg *sync.WaitGroup) {
    defer wg.Done()

    log.Printf("connecting to %s", u)
    c, _, err := websocket.DefaultDialer.Dial(u, nil)
    if err != nil {
        log.Fatal("dial:", err)
    }
    defer c.Close()

    done := make(chan struct{})

    go func() {
        defer close(done)
        for {
            _, message, err := c.ReadMessage()
            if err != nil {
                log.Println("read:", err)
                return
            }
            log.Printf("recv: %s", message)
        }
    }()

    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-done:
            return
        case t := <-ticker.C:
            err := c.WriteMessage(websocket.TextMessage, []byte(t.String()))
            if err != nil {
                log.Println("write:", err)
                return
            }
        case <-shutdown:
            // Cleanly close the connection by sending a close message and then
            // waiting (with timeout) for the server to close the connection.
            err := c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
            if err != nil {
                log.Println("write close:", err)
                return
            }
            select {
            case <-done:
            case <-time.After(time.Second):
            }
            return
        }
    }
}

在 main() 中声明一个 sync.WaitGroup。对于要连接到的每个 websocket 端点，递增 WaitGroup 并启动 goroutine 以连接该端点。启动 goroutines 后，在 WaitGroup 等待 goroutines 完成。

var wg sync.WaitGroup
for _, u := range endpoints { // endpoints is []string 
                              // where elements are URLs 
                              // of endpoints to connect to.
    wg.Add(1)
    go connect(u, shutdown, &wg)
}
wg.Wait()

上面的代码经过编辑使其运行针对 Gorilla 的 echo 示例服务器 posted on the playground。

Golang - 为到不同服务器的多个连接扩展 websocket 客户端

Golang - Scaling a websocket client for multiple connections to different servers

sockets

scaling

go