我正在實現(xiàn)一個服務(wù)器來流式傳輸許多浮點數(shù)數(shù)組。我需要一些幫助來設(shè)計我的系統(tǒng)以實現(xiàn)以下目標：音頻進程必須是獨立的，即使沒有任何請求進來也能完成它的工作。我目前的方法使 DataProcess 函數(shù)等待直到有請求。因為通道只能給1個請求提供數(shù)據(jù)，2個以上的請求怎么才能拿到我在DataProcess中準備好的數(shù)據(jù)呢？為了實際流式傳輸數(shù)據(jù)，請求處理程序不能等待整個 DataProcess 完成，無論如何，一旦我們完成 DataProcess 中的每個迭代，就給處理程序數(shù)據(jù)？任何答復(fù)表示贊賞。這是我目前的想法：package mainimport ("fmt""io""net/http""strconv""time")func main() { c := AudioProcess() handleHello := makeHello(c) http.HandleFunc("/", handleHello) http.ListenAndServe(":8000", nil)}func makeHello(c chan string) func(http.ResponseWriter, *http.Request) { return func(w http.ResponseWriter, r *http.Request) { for item := range c { // this loop runs when channel c is closed io.WriteString(w, item) } }}func AudioProcess() chan string { c := make(chan string) go func() { for i := 0; i <= 10; i++ { // Iterate the audio file c <- strconv.Itoa(i) // have my frame of samples, send to channel c time.Sleep(time.Second) fmt.Println("send ", i) // logging } close(c) // done processing, close channel c }() return c }
1 回答
偶然的你
TA貢獻1841條經(jīng)驗 獲得超3個贊
我不完全確定這是否能解決您的問題，因為我不完全了解您的用例，但是，我已經(jīng)在下面提出了一個解決方案。
我已經(jīng)將 Gin 用于 HTTP 路由器，因為它對我來說更舒服，但我很確定您可以調(diào)整代碼以適合您的。我這樣做很匆忙（抱歉），所以可能存在我不知道的問題，但如果有任何問題，請告訴我。
簡而言之：
我創(chuàng)建了一個
Manager可以處理多個Client. 它還包含一個sync.Mutex以確保在任何給定時間只有一個線程正在修改clients；有一個
InitBackgroundTask()會生成一個隨機float64數(shù)，并將其傳遞給 ALLclients中的一個Manager（如果有的話）。如果沒有clients，我們就睡覺，然后繼續(xù)……索引處理程序處理添加和刪除客戶端?？蛻舳送ㄟ^ UUID 進行識別；
現(xiàn)在可能會發(fā)生 3 件事。客戶端在通過
<-c.Writer.CloseNotify()通道斷開連接時會自動刪除（因為該方法返回從而調(diào)用defer）。我們也可以float64在下一個后臺任務(wù)tick中接收隨機數(shù)。最后，如果我們在 20 秒內(nèi)沒有收到任何東西，我們也可以終止。
我在這里對您的需求做了幾個假設(shè)（例如，后臺任務(wù)將每 Y 分鐘返回一次 X）。如果您正在尋找更細粒度的流媒體，我建議您改用 websockets（并且仍然可以使用下面的模式）。
如果您有任何問題，請告訴我。
代碼：
package main
import (
"github.com/gin-gonic/gin"
"github.com/satori/go.uuid"
"log"
"math/rand"
"net/http"
"sync"
"time"
)
type Client struct {
uuid string
out chan float64
}
type Manager struct {
clients map[string]*Client
mutex sync.Mutex
}
func NewManager() *Manager {
m := new(Manager)
m.clients = make(map[string]*Client)
return m
}
func (m *Manager) AddClient(c *Client) {
m.mutex.Lock()
defer m.mutex.Unlock()
log.Printf("add client: %s\n", c.uuid)
m.clients[c.uuid] = c
}
func (m *Manager) DeleteClient(id string) {
m.mutex.Lock()
defer m.mutex.Unlock()
// log.Println("delete client: %s", c.uuid)
delete(m.clients, id)
}
func (m *Manager) InitBackgroundTask() {
for {
f64 := rand.Float64()
log.Printf("active clients: %d\n", len(m.clients))
for _, c := range m.clients {
c.out <- f64
}
log.Printf("sent output (%+v), sleeping for 10s...\n", f64)
time.Sleep(time.Second * 10)
}
}
func main() {
r := gin.Default()
m := NewManager()
go m.InitBackgroundTask()
r.GET("/", func(c *gin.Context) {
cl := new(Client)
cl.uuid = uuid.NewV4().String()
cl.out = make(chan float64)
defer m.DeleteClient(cl.uuid)
m.AddClient(cl)
select {
case <-c.Writer.CloseNotify():
log.Printf("%s : disconnected\n", cl.uuid)
case out := <-cl.out:
log.Printf("%s : received %+v\n", out)
c.JSON(http.StatusOK, gin.H{
"output": out,
})
case <-time.After(time.Second * 20):
log.Println("timed out")
}
})
r.Run()
}
注意：如果您在 Chrome 上進行測試，您可能需要在 URL 的末尾附加一個隨機參數(shù)，以便實際發(fā)出請求，例如?rand=001，?rand=002等等。
- 1 回答
- 0 關(guān)注
- 115 瀏覽
添加回答
舉報
0/150
提交
取消