Приоритет в методе выбора выбора Go

Я хочу, чтобы рутинное прослушивание было прослушиванием на двух каналах, заблокировано, когда оба канала слиты. Однако, если оба канала содержат данные, я хочу, чтобы один из них был удален до того, как будет обработан другой.

В приведенном ниже рабочем примере я хочу, чтобы все out были разряжены до обработки exit. Я использую select -statement, который не имеет приоритетного порядка. Как я могу обойти проблему, заставляя все 10 исходных значений обрабатываться до выхода?

package main

import "fmt"

func sender(out chan int, exit chan bool){
    for i := 1; i <= 10; i++ {
        out <- i
    } 
    exit <- true
}

func main(){
    out := make(chan int, 10)
    exit := make(chan bool)

    go sender(out, exit)

    L:
    for {
        select {
            case i := <-out:
                fmt.Printf("Value: %d\n", i)
            case <-exit:
                fmt.Println("Exiting")
                break L
        }
    }
    fmt.Println("Did we get all 10? Most likely not")
}

Ответ 1

package main

import "fmt"

func sender(out chan int, exit chan bool) {
    for i := 1; i <= 10; i++ {
        out <- i
    }
    exit <- true
}

func main() {
    out := make(chan int, 10)
    exit := make(chan bool)

    go sender(out, exit)

    for {
        select {
        case i := <-out:
            fmt.Printf("Value: %d\n", i)
            continue
        default:
        }
        select {
        case i := <-out:
            fmt.Printf("Value: %d\n", i)
            continue
        case <-exit:
            fmt.Println("Exiting")
        }
        break
    }
    fmt.Println("Did we get all 10? I think so!")
}

Случай первого выбора по умолчанию делает его неблокирующим. Выбор вытолкнет выходной канал, не глядя на выходной канал, но в противном случае не будет ждать. Если выходной канал пуст, он сразу же падает до второго выбора. Второй выбор блокирует. Он будет ожидать данные на любом канале. Если выход завершен, он обрабатывает его и позволяет выйти из цикла. Если данные поступают, он возвращается в верхнюю часть цикла и возвращается в режим слива.

Ответ 2

язык поддерживает это изначально, и обходной путь не требуется. Это очень просто: канал выхода должен быть видимым только для производителя. Когда выходите, продюсер закрывает канал. Только когда канал пуст и закрыт, потребитель уходит. Это стало возможным благодаря чтению с канала следующим образом:

v, ok := <-c

это установит ok в логическое значение, указывающее, действительно ли значение v было считано с канала (ok == true), или если v было установлено нулевое значение типа, обрабатываемого канал c, потому что c закрыт и пуст (ok == false). Когда канал закрыт и не пуст, v будет действительным значением, а ok будет true. Когда канал закрыт и пуст, v будет нулевым значением типа, обрабатываемого каналом c, а ok будет false, что означает, что v бесполезен.

Вот пример, иллюстрирующий:

package main

import (
    "fmt"
    "math/rand"
    "time"
)

var (
    produced  = 0
    processed = 0
)

func produceEndlessly(out chan int, quit chan bool) {
    defer close(out)
    for {
        select {
        case <-quit:
            fmt.Println("RECV QUIT")
            return
        default:
            out <- rand.Int()
            time.Sleep(time.Duration(rand.Int63n(5e6)))
            produced++
        }
    }
}

func quitRandomly(quit chan bool) {
    d := time.Duration(rand.Int63n(5e9))
    fmt.Println("SLEEP", d)
    time.Sleep(d)
    fmt.Println("SEND QUIT")
    quit <- true
}

func main() {
    vals, quit := make(chan int, 10), make(chan bool)
    go produceEndlessly(vals, quit)
    go quitRandomly(quit)
    for {
        x, ok := <-vals
        if !ok {
            break
        }
        fmt.Println(x)
        processed++
        time.Sleep(time.Duration(rand.Int63n(5e8)))
    }
    fmt.Println("Produced:", produced)
    fmt.Println("Processed:", processed)
}

это описано в разделе "Оператор приема" спецификации go: http://golang.org/ref/spec#Receive_operator

Ответ 3

Другой подход:

package main

import "fmt"

func sender(c chan int) chan int {
        go func() {
                for i := 1; i <= 15; i++ {
                        c <- i
                }
                close(c)
        }()
        return c
}

func main() {
        for i := range sender(make(chan int, 10)) {
                fmt.Printf("Value: %d\n", i)
        }
        fmt.Println("Did we get all 15? Surely yes")
}

$ go run main.go
Value: 1
Value: 2
Value: 3
Value: 4
Value: 5
Value: 6
Value: 7
Value: 8
Value: 9
Value: 10
Value: 11
Value: 12
Value: 13
Value: 14
Value: 15
Did we get all 15? Surely yes
$

Ответ 4

Я создал одно довольно простое решение. Он делает то, что я хочу, но если у кого-то еще есть лучшее решение, сообщите мне:

exiting := false
for !exiting || len(out)>0 {
    select {
        case i := <-out:
            fmt.Printf("Value: %d\n", i)
        case <-exit:
            exiting = true
            fmt.Println("Exiting")
    }
}

Вместо того, чтобы выйти из приема, я отмечаю выход, выходящий, как только я убедился, что в chan out ничего не осталось.

Ответ 5

В моем случае я действительно хотел перенести данные из одного канала на другой, а не просто иметь внеполосный сигнал выхода. В интересах кого-то еще с той же проблемой, я думаю, что этот подход работает без потенциального состояния гонки:

OUTER:
for channelA != nil || channelB != nil {

    select {

    case typeA, ok := <-channelA:
        if !ok {
            channelA = nil
            continue OUTER
        }
        doSomething(typeA)

    case nodeIn, ok := <-channelB:
        if !ok {
            channelB = nil
            continue OUTER
        }

        // Looped non-blocking nested select here checks that channelA
        // really is drained before we deal with the data from channelB
        NESTED:
        for {
            select {
            case typeA, ok := <-channelA:
                if !ok {
                    channelA = nil
                    continue NESTED
                }
                doSomething(typeA)

            default:
                // We are free to process the typeB data now
                doSomethingElse(typeB)
                break NESTED
            }
        }
    }

}

Ответ 6

Я думаю, что ответ Соня неверен. Это мое решение, немного усложняющее.

package main

import "fmt"

func sender(out chan int, exit chan bool){
    for i := 1; i <= 10; i++ {
        out <- i
    } 
    exit <- true
}

func main(){
    out := make(chan int, 10)
    exit := make(chan bool)

    go sender(out, exit)

    L:
    for {
        select {
            case i := <-out:
                fmt.Printf("Value: %d\n", i)
            case <-exit:
                for{
                    select{
                    case i:=<-out:
                        fmt.Printf("Value: %d\n", i)
                    default:
                        fmt.Println("Exiting")
                        break L
                    }
                }
                fmt.Println("Exiting")
                break L
        }
    }
    fmt.Println("Did we get all 10? Yes!")
}

Ответ 7

Есть ли какая-либо конкретная причина для использования буферизованного канала make(chan int, 10)?

Вам нужно использовать небуферизованный канал с буферизацией, который вы используете.

Просто удалите 10, это должно быть просто make(chan int).

Таким образом выполнение в функции sender может перейти только к оператору exit <- true после того, как последнее сообщение из канала out будет отменено оператором i := <-out. Если это утверждение не было выполнено, в goroutine не может быть достигнуто exit <- true.

Ответ 8

Вот еще один вариант.

Код потребителя:

  go func() {
    stop := false
    for {
      select {
      case item, _ := <-r.queue:
        doWork(item)
      case <-r.stopping:
        stop = true
      }
      if stop && len(r.queue) == 0 {
        break
      }
    }
  }()