我想做一個動態通道池，它可以監聽數十萬個通道，所有這些都在控制之下，正如我所排除的那樣，我希望它能夠自動升級，如果有太多的通道監聽（goroutine =>反射=>選擇）但是在selectN通道觀察器編碼期間，我被通道替換阻止了我想在運行時替換chan，這是在選擇循環中，我已經嘗試了一段時間以使其可用，但事情進展不順利。func Test_Change(t *testing.T) {type A struct { ch chan interface{}}a := &A{ ch: make(chan interface{}),}go func() { for { select { case v := <-a.ch: fmt.Println(v) } }}()newCh := make(chan interface{})go func() { for i := 0; i < 200; i++ { a.ch <- i } a.ch = newCh}()go func() { for i := 1000; i < 1010; i++ { newCh <- i }}()for { select {}}}它阻止了func Test_Change(t *testing.T) {type A struct { ch chan interface{} bh chan interface{}}a := &A{ ch: make(chan interface{}), bh: make(chan interface{}),}notify := make(chan struct{})go func() { for { select { case v := <-a.ch: fmt.Println(v) case <-notify: fmt.Println("notify") } }}()newCh := make(chan interface{})go func() { for i := 0; i < 200; i++ { a.ch <- i } a.ch = newCh notify <- struct{}{}}()go func() { for i := 1000; i < 1010; i++ { newCh <- i }}()for { select {}}}它奏效了
3 回答
慕斯709654
TA貢獻1840條經驗 獲得超5個贊
您已經正確確定，使用常規語法（固定數量的事例）不可能在任意大的動態 chan 上設置一個塊，并且可以使用反射包。select
但是，我不確定這是實現目標的最佳方法。如果您確實有數千個頻道需要觀看（例如，同時連接數千個遠程客戶端），則可以使用“扇入”模式將所有內容寫入非常少的固定數量的頻道，并選擇該頻道。
而不是
for {
select {
case <-sigterm:
cleanup()
os.Exit(1)
case msg := <-client1:
// process msg...
case msg := <-client2:
// process msg...
// HOW CAN I DYNAMICALLY ADD AND REMOVE A CLIENT HERE?
}
}
想想像這樣：
for {
select {
case <-sigterm:
cleanup()
os.Exit(1)
case msg := <-clients:
// process msg...
}
}
func addClient(client chan Message) {
// Fan-in: read all future messages from client, and write them
// to clients.
go func(){
for msg := range client {
clients <- msg
}
}()
}
替換通道變量的值不是線程安全的（可以是數據爭用），但是讓多個戈魯廷同時寫入和讀取同一通道是完全可以的。clients
瀟湘沐
TA貢獻1816條經驗 獲得超6個贊
數據競賽是一個嚴重的錯誤和設計缺陷。您可以通過 使用 運行測試或使用 運行程序來檢測數據爭用。a.chgo test -racego run -race program.go
可以替換在 for/select 循環中使用的 chan 的值，只要它是在事例的主體內正確完成的，而不是在另一個并發 goroutine 的代碼中。
replace := time.After(3 * time.Second)
for {
select {
case v, ok := <-ch1:
// use v...
case v, ok := <-ch2:
// use v...
case <-replace:
ch1 = anotherChannel
}
}
此示例可運行代碼不雅（不要這樣做）。您可以將其保存在工作站上，然后嘗試使用數據競速檢測器。
此固定的示例代碼不具有不雅性。
炎炎設計
TA貢獻1808條經驗 獲得超4個贊
也許這可能有效;
非常非常廣泛地測試它，它是可行的，但很難把它弄對。
我預計它不會是完美的。而且它缺乏關于“該做”和“不該做”的大量文檔。
它也不是合并輸入通道的版本，它總是一次只消耗一個輸入通道。這可能會對性能造成問題。
我給出的唯一保證是它沒有比賽。
雖然我把寫一個同人版的任務留給讀者作為練習。
package main
import (
"fmt"
)
func main() {
m := New()
go m.Run()
input := m.Resize(0)
input <- 5
input <- 4
close(input)
input = m.Resize(10)
input <- 6
input <- 7
close(input)
input = m.Resize(2)
input <- 8
input <- 9
close(input)
m.Close()
fmt.Println()
}
type masterOfThings struct {
notify chan notification
wantClose chan chan bool
}
func New() masterOfThings {
return masterOfThings{
notify: make(chan notification, 1),
wantClose: make(chan chan bool),
}
}
type notification struct {
N int
out chan chan interface{}
}
func (m masterOfThings) Resize(n int) chan<- interface{} {
N := notification{
N: n,
out: make(chan chan interface{}, 1),
}
m.notify <- N
return <-N.out
}
func (m masterOfThings) Close() {
closed := make(chan bool)
m.wantClose <- closed
<-closed
}
func (m masterOfThings) Run() {
var input chan interface{}
inputs := []chan interface{}{}
closers := []chan bool{}
defer func() {
for _, c := range closers {
close(c)
}
}()
var wantClose bool
for {
select {
case m := <-m.wantClose:
closers = append(closers, m)
wantClose = true
if len(inputs) < 1 && input == nil {
return
}
case n, ok := <-input:
if !ok {
input = nil
if len(inputs) > 0 {
input = inputs[0]
copy(inputs, inputs[1:])
inputs = inputs[:len(inputs)-1]
} else if wantClose {
return
}
continue
}
fmt.Println(n)
case n := <-m.notify:
nInput := make(chan interface{}, n.N)
if input == nil {
input = nInput
} else {
inputs = append(inputs, nInput)
}
n.out <- nInput
}
}
}
- 3 回答
- 0 關注
- 143 瀏覽
添加回答
舉報
0/150
提交
取消