我從 Go 開始，現在我正在編寫一個簡單的程序，它從傳感器中讀取數據并將其放入通道中以使用它進行一些計算。我現在讓它工作如下：package mainimport ( "fmt" "time" "strconv")func get_sensor_data(c chan float64) { time.Sleep(1 * time.Second) // wait a second before sensor data starts pooring in c <- 2.1 // Sensor data starts being generated c <- 2.2 c <- 2.3 c <- 2.4 c <- 2.5}func main() { s := 1.1 c := make(chan float64) go get_sensor_data(c) for { select { case s = <-c: fmt.Println("the next value of s from the channel: " + strconv.FormatFloat(s, 'f', 1, 64)) default: // no new values in the channel } fmt.Println(s) time.Sleep(500 * time.Millisecond) // Do heavy "work" }}這很好用，但傳感器會生成大量數據，而我總是只對最新數據感興趣。然而，使用此設置，它只會在每個循環中讀出下一個項目，這意味著如果某個點的通道包含 20 個值，則只會在 10 秒后讀出最新值。有沒有辦法讓一個通道一次總是只包含一個值，這樣我總是只得到我感興趣的數據，并且通道沒有使用不必要的內存（盡管內存是我最不擔心的)?
5 回答
繁華開滿天機
TA貢獻1816條經驗 獲得超4個贊
最好將通道視為隊列 (FIFO)。因此你不能真的跳過。然而，有些庫可以做這樣的事情：https://github.com/cloudfoundry/go-diodes是一個原子環形緩沖區，它將覆蓋舊數據。如果您愿意，可以設置較小的尺寸。
綜上所述，聽起來您并不需要隊列（或環形緩沖區）。你只需要一個互斥量：
type SensorData struct{
? mu sync.RWMutex
? last float64
}
func (d *SensorData) Store(data float64) {
?mu.Lock()
?defer mu.Unlock()
?d.last = data
}
func (d *SensorData) Get() float64 {
?mu.RLock()
?defer mu.RUnlock()
?return d.last
}
這使用 aRWMutex這意味著許多東西可以同時從中讀取，而只有一個東西可以寫入。它將像您所說的那樣存儲一個條目。
慕妹3242003
TA貢獻1824條經驗 獲得超6個贊
不，通道是 FIFO 緩沖區，句號。這就是渠道的運作方式及其唯一目的。如果您只想要最新的值，請考慮只使用一個受互斥鎖保護的變量；每當有新數據進入時寫入它，無論何時讀取它，您將始終讀取最新值。
拉丁的傳說
TA貢獻1789條經驗 獲得超8個贊
你不能直接從一個通道獲取它，但你可以為每個值使用一個通道，并在有新值時得到通知：
package main
import (
? ? "fmt"
? ? "strconv"
? ? "sync"
? ? "time"
)
type LatestChannel struct {
? ? n? ? float64
? ? next chan struct{}
? ? mu? ?sync.Mutex
}
func New() *LatestChannel {
? ? return &LatestChannel{next: make(chan struct{})}
}
func (c *LatestChannel) Push(n float64) {
? ? c.mu.Lock()
? ? c.n = n
? ? old := c.next
? ? c.next = make(chan struct{})
? ? c.mu.Unlock()
? ? close(old)
}
func (c *LatestChannel) Get() (float64, <-chan struct{}) {
? ? c.mu.Lock()
? ? n := c.n
? ? next := c.next
? ? c.mu.Unlock()
? ? return n, next
}
func getSensorData(c *LatestChannel) {
? ? time.Sleep(1 * time.Second)
? ? c.Push(2.1)
? ? time.Sleep(100 * time.Millisecond)
? ? c.Push(2.2)
? ? time.Sleep(100 * time.Millisecond)
? ? c.Push(2.3)
? ? time.Sleep(100 * time.Millisecond)
? ? c.Push(2.4)
? ? time.Sleep(100 * time.Millisecond)
? ? c.Push(2.5)
}
func main() {
? ? s := 1.1
? ? c := New()
? ? _, hasNext := c.Get()
? ? go getSensorData(c)
? ? for {
? ? ? ? select {
? ? ? ? case <-hasNext:
? ? ? ? ? ? s, hasNext = c.Get()
? ? ? ? ? ? fmt.Println("the next value of s from the channel: " + strconv.FormatFloat(s, 'f', 1, 64))
? ? ? ? default:
? ? ? ? ? ? // no new values in the channel
? ? ? ? }
? ? ? ? fmt.Println(s)
? ? ? ? time.Sleep(250 * time.Millisecond) // Do heavy "work"
? ? }
}
心有法竹
TA貢獻1866條經驗 獲得超5個贊
試試這個包https://github.com/subbuv26/chanup
它允許生產者用最新值更新通道，從而替換最新值。并且產品不會被阻止。（有了這個，過時的值被覆蓋）。因此，在消費者方面，總是只讀取最新的項目。
import "github.com/subbuv26/chanup"
ch := chanup.GetChan()
_ := ch.Put(testType{
? ? a: 10,
? ? s: "Sample",
})
_ := ch.Update(testType{
? ? a: 20,
? ? s: "Sample2",
})
// Continue updating with latest values
...
...
// On consumer end
val := ch.Get()
// val contains latest value
- 5 回答
- 0 關注
- 183 瀏覽
添加回答
舉報
0/150
提交
取消