执行：减少等待组计数

偷偷努力，悄无声息地变强，然后惊艳所有人！哈哈，小伙伴们又来学习啦~今天我将给大家介绍《执行：减少等待组计数》，这篇文章主要会讲到等等知识点，不知道大家对其都有多少了解，下面我们就一起来看一吧！当然，非常希望大家能多多评论，给出合理的建议，我们一起学习，一起进步！

我是个新手，正在重新编写在其他地方找到的代码以满足我的需求。因此，尽管我了解了总体思路，但我并不完全理解这里发生的事情。

我正在使用 go 例程运行一些 websocket 客户端，但遇到意外错误，导致程序崩溃。当从 websocket 读取消息时发生错误（检查 readhandler 函数中的 conn.readmessage() 函数）时，我的程序似乎关闭了太多线程（如果这是错误的术语，请原谅）。关于如何解决这个问题有什么想法吗？我真的很感激任何人花时间阅读它。预先感谢！

package main

import (
    "context"
    "fmt"
    "os"
    "time"
    "os/signal"
    "syscall"
    "sync"
    "net/url"
    "github.com/gorilla/websocket"
    "strconv"
    "encoding/json"
    "log"
    "bytes"
    "compress/gzip"
    "io/ioutil"
)

// Structs

type Ping struct {
    Ping    int64   `json:"ping"`
}

type Pong struct {
    Pong        int64       `json:"pong"`
}

type SubParams struct {
    Sub         string          `json:"sub"`
    ID          string          `json:"id"`
}

func InitSub(subType string, pair string, i int) []byte {
    var idInt string = "id" + strconv.Itoa(i)
    subStr := "market." + pair + "." + subType
    sub := &SubParams{
        Sub: subStr,
        ID: idInt,
    }

    out, err := json.MarshalIndent(sub, "", " ")
    if err != nil {
        log.Println(err);
    }
    //log.Println(string(out))
    return out
}

// main func

func main() {
    var server string = "api.huobi.pro"
    pairs := []string{"btcusdt", "ethusdt", "ltcusdt"}
    comms := make(chan os.Signal, 1)
    signal.Notify(comms, os.Interrupt, syscall.SIGTERM)

    ctx := context.Background()
    ctx, cancel := context.WithCancel(ctx)
    var wg sync.WaitGroup

    for x, pair := range pairs {
        wg.Add(1)
        go control(server, "ws", pair, ctx, &wg, x+1)
    }

    <-comms
    cancel()
    wg.Wait()
}

func control(server string, path string, pair string, ctx context.Context, wg *sync.WaitGroup, i int) {
    fmt.Printf("Started control for %s\n", server)
    url := url.URL {
        Scheme: "wss",
        Host: server,
        Path: path,
    }

    fmt.Println(url.String())

    conn, _, err := websocket.DefaultDialer.Dial(url.String(), nil)
    if err != nil {
        panic(err)
    }
    subscribe(conn, pair, i)
    defer conn.Close()

    var localwg sync.WaitGroup

    localwg.Add(1)
    go readHandler(ctx, conn, &localwg, server)

    <- ctx.Done()
    localwg.Wait()
    wg.Done()
    return
}

func readHandler(ctx context.Context, conn *websocket.Conn, wg *sync.WaitGroup, server string) {
    for {

        select {

            case <- ctx.Done():
                wg.Done()
                return
            default:
                _, p, err :=  conn.ReadMessage()
                if err != nil {
                    wg.Done()
                    fmt.Println(err)
                }
                r, err := gzip.NewReader(bytes.NewReader(p))
                if(err == nil) {
                    result, err := ioutil.ReadAll(r)
                    if(err != nil) {
                        fmt.Println(err)
                    }
                    d := string(result)
                    fmt.Println(d)

                    var ping Ping
                    json.Unmarshal([]byte(d), &ping)
                    if (ping.Ping > 0) {
                        str := Pong{Pong: ping.Ping}
                        msg, err := json.Marshal(str)
                        if (err == nil) {
                            fmt.Println(string(msg))
                            conn.WriteMessage(websocket.TextMessage, []byte(msg))
                        }
                    }
            }
        }
    }
}

func subscribe(conn *websocket.Conn, pair string, id int) {
    sub := string(InitSub("trade.detail", pair, id))

    err := conn.WriteMessage(websocket.TextMessage, []byte(sub))
    if err != nil {
        panic(err)
    }
}

正确答案

• 连接失败时跳出 readhandler 循环：

  _, p, err :=  conn.readmessage()
    if err != nil {
        wg.done()
        fmt.println(err)
        return // <--- add this line
    }

  如果没有返回，函数会在一个紧密的循环中旋转，读取错误，直到发生恐慌。

• 在 goroutine 的开头使用 defer wg.done() 以确保 done 被调用一次。

  func readHandler(ctx context.Context, conn *websocket.Conn, wg *sync.WaitGroup, server string) {
      defer wg.Done()
      for {
        select {
        case <-ctx.Done():
            return
        default:
            _, p, err := conn.ReadMessage()
            if err != nil {
                fmt.Println(err)
                return
            }
       ...

  同时更新 control 函数。

• 由于调用者不会与 readhander 同时执行任何代码，因此运行 readhandlerzqbreadhander 是一个 goroutine 没有任何价值。从 readhandler 中删除所有对等待组的引用，并直接调用该函数：将 go readhandler(ctx, conn, &localwg, server) 更改为 readhandler(ctx, conn, server)。

还有更多问题，但这应该会让您走得更远。

好了，本文到此结束，带大家了解了《执行：减少等待组计数》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！