Go: أنماط «Go Select» والتزامن

// Bad code: busy polling
func pollDataSources() {
    for {
        // Poll every 10ms, wasting CPU
        if data1 := pollNYSE(); data1 != nil {
            process(data1)
        }
        if data2 := pollNASDAQ(); data2 != nil {
            process(data2)
        }
        if data3 := pollLSE(); data3 != nil {
            process(data3)
        }
        time.Sleep(10 * time.Millisecond)
    }
}

// market_data.go
package main

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

type Quote struct {
    Source string
    Symbol string
    Price  float64
}

func simulateExchange(name string, out chan<- Quote) {
    symbols := []string{"AAPL", "GOOGL", "MSFT", "AMZN"}
    for {
        time.Sleep(time.Duration(50+rand.Intn(200)) * time.Millisecond)
        quote := Quote{
            Source: name,
            Symbol: symbols[rand.Intn(len(symbols))],
            Price:  100 + rand.Float64()*200,
        }
        out <- quote
    }
}

func main() {
    nyse := make(chan Quote)
    nasdaq := make(chan Quote)
    lse := make(chan Quote)

    go simulateExchange("NYSE", nyse)
    go simulateExchange("NASDAQ", nasdaq)
    go simulateExchange("LSE", lse)

    // الاختيار listens to all three channels simultaneously
    timeout := time.After(2 * time.Second)
    for {
        الاختيار {
        case q := <-nyse:
            fmt.Printf("[NYSE] %s: $%.2f\n", q.Symbol, q.Price)
        case q := <-nasdaq:
            fmt.Printf("[NASDAQ] %s: $%.2f\n", q.Symbol, q.Price)
        case q := <-lse:
            fmt.Printf("[LSE] %s: $%.2f\n", q.Symbol, q.Price)
        case <-timeout:
            fmt.Println("Demo ended")
            return
        }
    }
}

[NASDAQ] AMZN: $198.32
[NYSE] AAPL: $150.45
[LSE] MSFT: $287.10
...

select {
case v := <-ch1:
    // ch1 is ready
case v := <-ch2:
    // ch2 is ready
case ch3 <- value:
    // Can send to ch3 (ch3 has space or has a receiver)
default:
    // All channels are not ready
}

package main

import (
    "fmt"
)

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

    ch1 <- "from ch1"
    ch2 <- "from ch2"

    // Both channels are ready; الاختيار picks one at random
    for i := 0; i < 2; i++ {
        الاختيار {
        case msg := <-ch1:
            fmt.Println(msg)
        case msg := <-ch2:
            fmt.Println(msg)
        }
    }
}

from ch1
from ch2

package main

import (
    "fmt"
    "time"
)

func slowOperation() string {
    time.Sleep(2 * time.Second)
    return "result"
}

func main() {
    ch := make(chan string)
    go func() {
        ch <- slowOperation()
    }()

    select {
    case result := <-ch:
        fmt.Println("Success:", result)
    case <-time.After(1 * time.Second):
        fmt.Println("Timeout! Operation exceeded 1 second")
    }
}

Timeout! Operation exceeded 1 second

package main

import (
    "fmt"
    "time"
)

func fetchFromCache() string {
    time.Sleep(50 * time.Millisecond)
    return "cache-data"
}

func fetchFromDB() string {
    time.Sleep(200 * time.Millisecond)
    return "db-data"
}

func fetchFromAPI() string {
    time.Sleep(500 * time.Millisecond)
    return "api-data"
}

func main() {
    cache := make(chan string)
    db := make(chan string)
    api := make(chan string)

    go func() { cache <- fetchFromCache() }()
    go func() { db <- fetchFromDB() }()
    go func() { api <- fetchFromAPI() }()

    الاختيار {
    case r := <-cache:
        fmt.Println("Cache hit:", r)
    case <-time.After(100 * time.Millisecond):
        الاختيار {
        case r := <-db:
            fmt.Println("DB returned:", r)
        case <-time.After(300 * time.Millisecond):
            الاختيار {
            case r := <-api:
                fmt.Println("API returned:", r)
            case <-time.After(600 * time.Millisecond):
                fmt.Println("All data sources timed out!")
            }
        }
    }
}

package main

import (
    "fmt"
)

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

    // Non-blocking receive
    select {
    case v := <-ch:
        fmt.Println("Received:", v)
    default:
        fmt.Println("No data (non-blocking)")
    }

    // Non-blocking send
    ch <- 1
    select {
    case ch <- 2:
        fmt.Println("Send successful")
    default:
        fmt.Println("Buffer full (non-blocking)")
    }
}

No data (non-blocking)
Buffer full (non-blocking)

package main

import (
    "fmt"
    "time"
)

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

    go func() {
        for i := 1; i <= 10; i++ {
            select {
            case ch <- i:
                // Send successful
            default:
                fmt.Printf("Buffer full, dropped %d\n", i)
            }
            time.Sleep(10 * time.Millisecond)
        }
        close(ch)
    }()

    // Slow consumer
    for v := range ch {
        fmt.Printf("Processing: %d\n", v)
        time.Sleep(50 * time.Millisecond)
    }
}

Processing: 1
Processing: 2
Processing: 3
Buffer full, dropped 4
Buffer full, dropped 5
Processing: 6
...

package main

import (
    "fmt"
    "time"
)

func worker(done <-chan struct{}) {
    for {
        select {
        case <-done:
            fmt.Println("worker exiting")
            return
        default:
            fmt.Println("worker working...")
            time.Sleep(200 * time.Millisecond)
        }
    }
}

func main() {
    done := make(chan struct{})
    go worker(done)

    time.Sleep(1 * time.Second)
    close(done)
    time.Sleep(100 * time.Millisecond)
    fmt.Println("main exiting")
}

package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan int)
    done := make(chan struct{})

    // Producer
    go func() {
        for i := 1; i <= 5; i++ {
            ch <- i
            time.Sleep(100 * time.Millisecond)
        }
        close(ch)
    }()

    // Consumer (for-select loop)
    go func() {
        for {
            select {
            case v, ok := <-ch:
                if !ok {
                    fmt.Println("Style 1: channel closed exit")
                    close(done)
                    return
                }
                fmt.Printf("Processing: %d\n", v)
            case <-time.After(1 * time.Second):
                fmt.Println("Style 2: timeout exit")
                close(done)
                return
            }
        }
    }()

    <-done
    fmt.Println("main exiting")
}

package main

import (
    "fmt"
    "sync"
)

func fanOut(jobs <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)

    for w := 0; w < workers; w++ {
        ch := make(chan int, 10)
        channels[w] = ch

        go func(id int, out chan<- int) {
            defer close(out)
            for job := range jobs {
                result := job * job
                fmt.Printf("Worker %d: %d^2 = %d\n", id, job, result)
                out <- result
            }
        }(w, ch)
    }

    return channels
}

package main

import (
    "fmt"
    "sync"
)

func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    jobs := make(chan int, 10)
    for i := 1; i <= 6; i++ {
        jobs <- i
    }
    close(jobs)

    // fan-out: distribute to 3 workers
    workers := fanOut(jobs, 3)

    // fan-in: aggregate all worker results
    results := fanIn(workers...)

    // Collect results
    sum := 0
    count := 0
    for r := range results {
        sum += r
        count++
    }
    fmt.Printf("Total %d results, sum = %d\n", count, sum)
}

Worker 2: 1^2 = 1
Worker 0: 3^2 = 9
Worker 1: 2^2 = 4
Worker 1: 5^2 = 25
Worker 0: 4^2 = 16
Worker 2: 6^2 = 36
Total 6 results, sum = 91

// market_aggregator.go
package main

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

// ---------- Data Types ----------

type Trade struct {
    Source string
    Symbol string
    Price  float64
    Volume int
    Time   time.Time
}

type AggregatedQuote struct {
    Symbol      string
    AvgPrice    float64
    TotalVolume int
    Sources     int
    High        float64
    Low         float64
    Time        time.Time
}

// ---------- Simulated Exchange Data Source ----------

var symbols = []string{"AAPL", "GOOGL", "MSFT", "AMZN", "TSLA", "META"}

func simulateExchange(name string, out chan<- Trade, done <-chan struct{}) {
    for {
        select {
        case <-done:
            return
        default:
            time.Sleep(time.Duration(100+rand.Intn(300)) * time.Millisecond)
            trade := Trade{
                Source: name,
                Symbol: symbols[rand.Intn(len(symbols))],
                Price:  100 + rand.Float64()*200,
                Volume: rand.Intn(1000) + 100,
                Time:   time.Now(),
            }
            select {
            case out <- trade:
            case <-done:
                return
            }
        }
    }
}

// ---------- Aggregator ----------

type Aggregator struct {
    trades chan Trade
    done   chan struct{}
    wg     sync.WaitGroup
}

func NewAggregator() *Aggregator {
    return &Aggregator{
        trades: make(chan Trade, 100),
        done:   make(chan struct{}),
    }
}

func (a *Aggregator) AddSource(name string) {
    a.wg.Add(1)
    go func() {
        defer a.wg.Done()
        simulateExchange(name, a.trades, a.done)
    }()
    fmt.Printf("Added data source: %s\n", name)
}

func (a *Aggregator) Start(interval time.Duration, callback func(map[string]*AggregatedQuote)) {
    // Aggregation window
    window := make(map[string][]Trade)

    ticker := time.NewTicker(interval)
    defer ticker.Stop()

    for {
        select {
        case trade := <-a.trades:
            window[trade.Symbol] = append(window[trade.Symbol], trade)

        case <-ticker.C:
            // Time window reached; compute aggregated quotes
            quotes := make(map[string]*AggregatedQuote)

            for symbol, trades := range window {
                if len(trades) == 0 {
                    continue
                }
                var sumPrice float64
                var totalVol int
                high := trades[0].Price
                low := trades[0].Price

                for _, t := range trades {
                    sumPrice += t.Price * float64(t.Volume)
                    totalVol += t.Volume
                    if t.Price > high {
                        high = t.Price
                    }
                    if t.Price < low {
                        low = t.Price
                    }
                }

                sources := make(map[string]bool)
                for _, t := range trades {
                    sources[t.Source] = true
                }

                quotes[symbol] = &AggregatedQuote{
                    Symbol:      symbol,
                    AvgPrice:    sumPrice / float64(totalVol),
                    TotalVolume: totalVol,
                    Sources:     len(sources),
                    High:        high,
                    Low:         low,
                    Time:        time.Now(),
                }
            }

            callback(quotes)
            window = make(map[string][]Trade) // Reset window

        case <-a.done:
            return
        }
    }
}

func (a *Aggregator) Stop() {
    close(a.done)
    a.wg.Wait()
}

// ---------- Main Function ----------

func main() {
    aggregator := NewAggregator()

    // Add 3 data sources
    aggregator.AddSource("NYSE")
    aggregator.AddSource("NASDAQ")
    aggregator.AddSource("LSE")

    fmt.Println("\nStarting aggregation (output every 2 seconds)...\n")

    // Start aggregation with 2-second time window
    done := make(chan struct{})
    go func() {
        aggregator.Start(2*time.Second, func(quotes map[string]*AggregatedQuote) {
            fmt.Printf("=== Aggregation Report %s ===\n", time.Now().Format("15:04:05"))
            for _, q := range quotes {
                fmt.Printf("%-6s | $%.2f (avg) | vol=%d | src=%d | H=%.2f L=%.2f\n",
                    q.Symbol, q.AvgPrice, q.TotalVolume, q.Sources, q.High, q.Low)
            }
            fmt.Println()
        })
        close(done)
    }()

    // Run for 10 seconds then stop
    time.Sleep(10 * time.Second)
    aggregator.Stop()
    <-done
    fmt.Println("Quote aggregator stopped")
}

Added data source: NYSE
Added data source: NASDAQ
Added data source: LSE

Starting aggregation (output every 2 seconds)...

=== Aggregation Report 10:00:02 ===
AAPL   | $152.34 (avg) | vol=2341 | src=3 | H=165.20 L=142.10
GOOGL  | $178.90 (avg) | vol=1567 | src=2 | H=185.00 L=172.30
MSFT   | $295.40 (avg) | vol=890  | src=3 | H=301.20 L=288.50

=== Aggregation Report 10:00:04 ===
...

Quote aggregator stopped