Каналы

Каналы (Channels)

Что такое канал?

Канал (channel) -- это типизированный механизм коммуникации между горутинами. Каналы обеспечивают синхронизацию и передачу данных одновременно. Это реализация принципа Go: "Не общайтесь через разделяемую память -- разделяйте память через общение" (Share memory by communicating).

Горутина A ──► [  канал  ] ──► Горутина B
               ch <- val        val := <-ch

Каналы -- это first-class values в Go: их можно передавать как аргументы, возвращать из функций, хранить в структурах.

Создание каналов

package main

import "fmt"

func main() {
    // Unbuffered channel — synchronous
    ch1 := make(chan int)

    // Buffered channel — async until buffer is full
    ch2 := make(chan string, 10)

    // Channel of channels — meta pattern
    ch3 := make(chan chan int)

    // Channel of structs — for signaling (zero-size)
    done := make(chan struct{})

    fmt.Printf("ch1: %T, ch2: %T, ch3: %T, done: %T\n", ch1, ch2, ch3, done)
}

Небуферизованные каналы (Unbuffered)

Небуферизованный канал не имеет внутренней ёмкости. Отправка и получение блокируют горутину до тех пор, пока обе стороны не будут готовы. Это делает передачу синхронной -- гарантируется, что данные переданы.

Горутина A:  ch <- 42   ──► БЛОК ─────────► продолжение
                                    │
Горутина B:          ... ──► <-ch ──┘ (обе разблокированы одновременно)

package main

import "fmt"

func main() {
    ch := make(chan int) // unbuffered

    go func() {
        // This blocks until someone reads from ch
        ch <- 42
        fmt.Println("Sent 42")
    }()

    // This blocks until someone writes to ch
    value := <-ch
    fmt.Println("Received:", value)
    // Output:
    // Received: 42
    // Sent 42
}

Ключевой момент: В небуферизованном канале отправитель блокируется до момента, когда получатель готов принять значение, и наоборот. Это создаёт точку синхронизации (happens-before guarantee).

Буферизованные каналы (Buffered)

Буферизованный канал имеет внутреннюю очередь FIFO. Отправка блокируется только когда буфер полон, а получение -- только когда буфер пуст.

Буфер размера 3:

ch <- 1   [1|_|_]  — не блокирует
ch <- 2   [1|2|_]  — не блокирует
ch <- 3   [1|2|3]  — не блокирует
ch <- 4   [1|2|3]  — БЛОКИРУЕТ (буфер полон)

<-ch      [2|3|_]  — возвращает 1, разблокирует отправителя

package main

import "fmt"

func main() {
    // Buffered channel with capacity 3
    ch := make(chan string, 3)

    // These don't block — buffer has space
    ch <- "first"
    ch <- "second"
    ch <- "third"

    fmt.Println("Buffer length:", len(ch)) // 3
    fmt.Println("Buffer capacity:", cap(ch)) // 3

    // Read in FIFO order
    fmt.Println(<-ch) // first
    fmt.Println(<-ch) // second
    fmt.Println(<-ch) // third
}

Когда использовать буферизованные каналы

Сценарий	Буфер
Синхронизация (handshake)	0 (unbuffered)
Известное число задач	N (количество задач)
Пул воркеров	N (количество воркеров)
Rate limiting	1 (семафор)
Вещание (fan-out)	0

Направленность каналов (Channel Direction)

Go позволяет ограничить направление канала в сигнатуре функции. Это обеспечивает безопасность на уровне типов.

package main

import "fmt"

// producer can ONLY send to channel
func producer(out chan<- int) {
    for i := range 5 {
        out <- i
    }
    close(out)
}

// consumer can ONLY receive from channel
func consumer(in <-chan int) {
    for val := range in {
        fmt.Printf("Got: %d\n", val)
    }
}

func main() {
    ch := make(chan int, 5) // bidirectional

    // Bidirectional channel is implicitly converted to directional
    go producer(ch) // chan int → chan<- int
    consumer(ch)    // chan int → <-chan int
}

Направления:

chan T -- двунаправленный (чтение и запись)
chan<- T -- только отправка (send-only)
<-chan T -- только получение (receive-only)

Правило: Двунаправленный канал можно передать туда, где ожидается однонаправленный. Обратное -- невозможно. Это используется для документирования и обеспечения контракта.

Закрытие каналов

Закрытие канала сигнализирует, что больше никаких значений отправлено не будет:

package main

import "fmt"

func main() {
    ch := make(chan int, 5)

    // Send values and close
    for i := 1; i <= 5; i++ {
        ch <- i
    }
    close(ch)

    // Method 1: range over channel (stops on close)
    for val := range ch {
        fmt.Println(val)
    }

    // Method 2: comma-ok idiom
    ch2 := make(chan int, 1)
    ch2 <- 42
    close(ch2)

    val, ok := <-ch2
    fmt.Printf("val=%d, ok=%v\n", val, ok) // val=42, ok=true

    val, ok = <-ch2
    fmt.Printf("val=%d, ok=%v\n", val, ok) // val=0, ok=false
}

Правила закрытия каналов:

Закрывать должен только отправитель, никогда получатель

Закрывать канал можно только один раз (повторное закрытие -- паника)

После закрытия чтение возвращает оставшиеся значения, затем нулевое значение типа

Запись в закрытый канал -- паника

Аксиомы каналов

Эти правила необходимо знать наизусть:

┌────────────────────────┬──────────────────────┬────────────────────────┐
│ Операция               │ nil канал            │ Закрытый канал         │
├────────────────────────┼──────────────────────┼────────────────────────┤
│ Отправка (ch <- v)     │ Блокирует навсегда   │ ПАНИКА                 │
│ Получение (<-ch)       │ Блокирует навсегда   │ Возвращает zero value  │
│ Закрытие (close(ch))   │ ПАНИКА               │ ПАНИКА                 │
│ len(ch)                │ 0                    │ Количество в буфере    │
│ cap(ch)                │ 0                    │ Ёмкость буфера         │
└────────────────────────┴──────────────────────┴────────────────────────┘

package main

func main() {
    // nil channel — blocks forever
    var ch chan int // nil

    // go func() { ch <- 1 }()  // blocks forever
    // go func() { <-ch }()      // blocks forever
    // close(ch)                  // panic: close of nil channel

    _ = ch
}

Важно: Свойство nil-канала блокировать навсегда активно используется в паттерне с select -- чтобы "отключить" один из case-ов, канал устанавливают в nil.

Оператор select

select позволяет горутине ожидать на нескольких операциях с каналами одновременно. Это мультиплексор для каналов.

package main

import (
    "fmt"
    "time"
)

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)

    go func() {
        time.Sleep(100 * time.Millisecond)
        ch1 <- "one"
    }()

    go func() {
        time.Sleep(200 * time.Millisecond)
        ch2 <- "two"
    }()

    // Wait for whichever channel is ready first
    for range 2 {
        select {
        case msg := <-ch1:
            fmt.Println("Received from ch1:", msg)
        case msg := <-ch2:
            fmt.Println("Received from ch2:", msg)
        }
    }
}

select с default (неблокирующий)

package main

import "fmt"

func main() {
    ch := make(chan int, 1)

    select {
    case val := <-ch:
        fmt.Println("Received:", val)
    default:
        // Executes immediately if no channel is ready
        fmt.Println("No data available")
    }
}

Паттерн timeout

package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan int)

    go func() {
        time.Sleep(3 * time.Second)
        ch <- 42
    }()

    select {
    case val := <-ch:
        fmt.Println("Received:", val)
    case <-time.After(1 * time.Second):
        fmt.Println("Timeout: no response in 1 second")
    }
}

Правила select

Если несколько case-ов готовы одновременно, выбирается случайный (pseudo-random)
Если ни один case не готов и есть default -- выполняется default
Если ни один case не готов и нет default -- горутина блокируется
select{} без case-ов блокирует навсегда (иногда используется в main)

Обнаружение дедлоков

Рантайм Go обнаруживает ситуации, когда все горутины заблокированы:

package main

func main() {
    ch := make(chan int)
    ch <- 1 // deadlock: main goroutine blocks, no other goroutines exist
}
// fatal error: all goroutines are asleep - deadlock!

package main

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)

    // Classic deadlock: A waits for B, B waits for A
    go func() {
        val := <-ch1 // waits for ch1
        ch2 <- val   // never reached
    }()

    val := <-ch2 // waits for ch2
    ch1 <- val   // never reached
}
// fatal error: all goroutines are asleep - deadlock!

Ограничение: Рантайм обнаруживает дедлок только когда все горутины заблокированы. Если хотя бы одна горутина работает (например, HTTP-сервер), дедлок среди других горутин не будет обнаружен.

Паттерн: Producer-Consumer

Классический паттерн "производитель-потребитель" с каналами:

package main

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

// Order represents a customer order
type Order struct {
    ID    int
    Item  string
    Price float64
}

// producer generates orders and sends them to the channel
func producer(orders chan<- Order, count int) {
    items := []string{"Pizza", "Burger", "Sushi", "Tacos", "Pasta"}
    for i := 1; i <= count; i++ {
        order := Order{
            ID:    i,
            Item:  items[i%len(items)],
            Price: float64(i) * 9.99,
        }
        orders <- order
        fmt.Printf("[Producer] Created order #%d: %s\n", order.ID, order.Item)
        time.Sleep(100 * time.Millisecond)
    }
    close(orders) // signal: no more orders
}

// consumer processes orders from the channel
func consumer(id int, orders <-chan Order, wg *sync.WaitGroup) {
    defer wg.Done()
    for order := range orders {
        fmt.Printf("[Consumer %d] Processing order #%d: %s ($%.2f)\n",
            id, order.ID, order.Item, order.Price)
        time.Sleep(250 * time.Millisecond) // simulate processing
    }
    fmt.Printf("[Consumer %d] No more orders, shutting down\n", id)
}

func main() {
    orders := make(chan Order, 5) // buffered channel

    // Start 1 producer
    go producer(orders, 10)

    // Start 3 consumers
    var wg sync.WaitGroup
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go consumer(i, orders, &wg)
    }

    wg.Wait()
    fmt.Println("All orders processed")
}

Паттерн: Пинг-понг (демонстрация синхронности)

package main

import (
    "fmt"
    "time"
)

func main() {
    ping := make(chan struct{})
    pong := make(chan struct{})

    // Player 1
    go func() {
        for range 5 {
            <-ping // wait for ball
            fmt.Println("Player 1: PING")
            time.Sleep(100 * time.Millisecond)
            pong <- struct{}{} // hit back
        }
    }()

    // Player 2
    go func() {
        for range 5 {
            <-pong // wait for ball
            fmt.Println("Player 2: PONG")
            time.Sleep(100 * time.Millisecond)
            ping <- struct{}{} // hit back
        }
    }()

    ping <- struct{}{} // serve the ball
    time.Sleep(2 * time.Second)
}

Идиома: done-канал

package main

import (
    "fmt"
    "time"
)

// doWork runs until done channel is closed
func doWork(done <-chan struct{}) <-chan int {
    results := make(chan int)
    go func() {
        defer close(results)
        i := 0
        for {
            select {
            case <-done:
                fmt.Println("Worker: received done signal")
                return
            case results <- i:
                i++
                time.Sleep(100 * time.Millisecond)
            }
        }
    }()
    return results
}

func main() {
    done := make(chan struct{})
    results := doWork(done)

    // Read 5 results
    for range 5 {
        fmt.Println("Result:", <-results)
    }

    // Signal worker to stop
    close(done)
    time.Sleep(200 * time.Millisecond)
    fmt.Println("Main: done")
}