我想监听多个传输编码响应的 HTTP 流，然后逐行从中获取消息，然后将消息推送到一个 channel 。然后我想从 channel 中读取并稍后通过 websocket 推送。

func subscribe(ws chan<- string, group string) (scanner *bufio.Scanner, err error){
    res, _ := req(STREAM_URL, channelTemplate(group))
    reader := bufio.NewScanner(res.Body)
    return reader, reader.Err()
}

func main() {
    ws := make(chan string)
    request, _ := http.NewRequest("GET", URL, nil)
    request.Header.Add("Content-Type", "application/json")
    client := &http.Client{}
    resp, _ := client.Do(request)
    ac := ACResponse{}
    json.NewDecoder(resp.Body).Decode(&ac)
    resp.Body.Close()
    var scanners = make([]*bufio.Scanner, 0)
    for _, group := range ac.Groups {
        fmt.Println("Started worker for", group)
        //listen to all stream URLs
        scanner, err := subscribe(ws, group)
        if err != nil {
            panic(err)
        }
        // keep track of Scanner to read later
        scanners = append(scanners, scanner)
    }
    for {
        select {
        case msg := <-ws:
            fmt.Println("[events] ", msg)
        default:
            randScanner := rand.Intn(len(ac.Groups)-1)
            fmt.Println("Reading from", randScanner)
            reader := scanners[randScanner]
            reader.Scan()
            if err := reader.Err(); err != nil {
                panic(err)
            }
            text := reader.Text()
            ws <- text
        }
    }
}

程序在 reader.Scan() 处阻塞。输出是 Reading from 1 而不是别的。我查看了 wireshark，消息正在通过。

如何使用 Go 更好地设计这个问题？

最佳答案

主要 block 发送到无缓冲 channel ws。要解决此问题，请将 ws 更改为缓冲 channel :

ws := make(chan string, 1)

第二个问题是 main() 在到达 EOF 后继续读取扫描仪。问题出在这些行上:

        reader.Scan()
        if err := reader.Err(); err != nil {
            panic(err)
        }
        text := reader.Text()

Scan() 在 EOF 处返回 false，但忽略 scan 的返回。 Err() 在 EOF 时返回 nil。修改应用程序以使用 Scan() 的返回值。

还有一个问题是任何一个扫描器读取时的主要 block 。为了避免在单个连接上阻塞，启动一个 goroutine 来读取每个连接:

func subscribe(wg *sync.WaitGroup, ws chan<- string, group string) {
    defer wg.Done()
    res, err := req(STREAM_URL, channelTemplate(group))
    if err ! nil {
         // handle error
    }
    defer resp.Body.Close()
    reader := bufio.NewScanner(res.Body)
    for reader.Scan() {
       ws <- reader.Text()
    }
    if err := reader.Err(); err != nil {
        // handle error
    }
}

func main() {
    ws := make(chan string)
    request, _ := http.NewRequest("GET", URL, nil)
    request.Header.Add("Content-Type", "application/json")
    resp, err := http.DefaultClient.Do(request)
    if err != nil {
         // handle error
    }
    var ac ACResponse
    if err := json.NewDecoder(resp.Body).Decode(&ac); err != nil {
       // handle error
    }
    resp.Body.Close()
    var wg sync.WaitGroup
    for _, group := range ac.Groups {
         wg.Add(1)
         go subscribe(&wg, ws, group)
    }

    go func() {
       wg.Wait()
       close(ws)
    }()

    for msg := range ws {
        fmt.Println("[events] ", msg)
    }
}

以上代码未经编译和测试。我已经标记了需要错误处理的地方。我编写了在所有连接到达 EOF 后退出 main 的代码。这可能是也可能不是您在应用程序中想要的。

关于go - HTTP Chunked 流式传输到 WebSocket，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/35350204/