go - 如何延迟直播？

Question

我正在尝试在 Go 中构建一项服务，该服务将直播流 (socketio/signalR) 延迟约 7 分钟。它还应该允许非延迟流。所以 Go 服务应该有类似缓冲区或队列的东西，强制数据等待指定的持续时间，然后才允许使用。你会如何在 Go 中做这样的事情？延迟流会是一个单独的 goroutine 吗？延迟数据应该用什么数据结构？

我目前的想法是使用 time 包等待/滴答 7 分钟，然后才允许使用数据，但这种阻塞行为在这种情况下可能不是最佳选择。

这是一些代码来解释我正在尝试做的事情。 FakeStream 是一个模拟函数，它模拟我从外部服务获取的实时流数据。

package main

import (
    "fmt"
    "time"
)

func DelayStream(input chan string, output chan string, delay string) {

    // not working for some reason
    // delayDuration, _ := time.ParseDuration(delay)
    // fmt.Println(delayDuration.Seconds())

    if delay == "5s" {
        fmt.Println("sleeping")
        time.Sleep(5 * time.Second)
    }
    data := <-input
    output <- data
}

func FakeStream(live chan string) {

    ticks := time.Tick(2 * time.Second)
    for now := range ticks {
        live <- fmt.Sprintf("%v", now.Format(time.UnixDate))
    }
}

func main() {
    liveData := make(chan string)
    delayedData := make(chan string)

    go FakeStream(liveData)
    go DelayStream(liveData, delayedData, "5s")

    for {
        select {
        case live := <-liveData:
            fmt.Println("live: ", live)
        case delayed := <-delayedData:
            fmt.Println("delayed: ", delayed)
        }
    }
}

出于某种原因，延迟的 channel 只输出一次，并没有输出预期的数据。它应该输出直播 channel 中的第一件事，但它没有。

Answer 1

您需要一个足够大的缓冲区。对于简单的情况，缓冲的 Go channel 可以工作。

问问自己 - 在此延迟期间有多少数据要存储 - 你应该有一个合理的上限。例如，如果您的流每秒传输多达 N 个数据包，那么要延迟 7 分钟，您将需要存储 420N 个数据包。

问问自己 - 如果在延迟窗口期间到达的数据多于预期会怎样？你可以扔掉新数据，或者扔掉旧数据，或者只是阻塞输入流。对于您的场景，哪些是可行的？每个都会导致略有不同的解决方案。

问问自己 - 延迟是如何计算的？从创建流的那一刻起？从每个数据包到达的那一刻起？每个数据包的延迟是单独的，还是只针对流中的第一个数据包？

为了开发一些示例代码，您需要大大缩小此处的设计选择范围。

对于这些设计选择的某些子集，这里有一个简单的方法来为每条消息添加 channel 之间的延迟:

---

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    // in is a channel of strings with a buffer size of 10
    in := make(chan string, 10)

    // out is an unbuffered channel
    out := make(chan string)

    // this goroutine forwards messages from in to out, ading a delay
    // to each message.
    const delay = 3 * time.Second
    go func() {
        for msg := range in {
            time.Sleep(delay)
            out <- msg
        }
        close(out)
    }()

    var wg sync.WaitGroup
    wg.Add(1)
    // this goroutine drains the out channel
    go func() {
        for msg := range out {
            fmt.Printf("Got '%s' at time %s\n", msg, time.Now().Format(time.Stamp))
        }
        wg.Done()
    }()

    // Send some messages into the in channel
    fmt.Printf("Sending '%s' at time %s\n", "joe", time.Now().Format(time.Stamp))
    in <- "joe"

    time.Sleep(2 * time.Second)
    fmt.Printf("Sending '%s' at time %s\n", "hello", time.Now().Format(time.Stamp))
    in <- "hello"

    time.Sleep(4 * time.Second)
    fmt.Printf("Sending '%s' at time %s\n", "bye", time.Now().Format(time.Stamp))
    in <- "bye"
    close(in)

    wg.Wait()
}