Goの高度な並行処理パターン

package main

import (
	"context"
	"fmt"
	"sync"
)

// Stage 1: Generator
func generator(ctx context.Context, nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			select {
			case out <- n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// Stage 2: Squarer
func squarer(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case out <- n * n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// Stage 3: Filter
func filterEven(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			if n%2 == 0 {
				select {
				case out <- n:
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	return out
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	nums := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
	genChan := generator(ctx, nums...)
	sqChan := squarer(ctx, genChan)
	filterChan := filterEven(ctx, sqChan)

	for val := range filterChan {
		fmt.Println(val)
	}
}

func merge(ctx context.Context, cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	output := func(c <-chan int) {
		defer wg.Done()
		for n := range c {
			select {
			case out <- n:
			case <-ctx.Done():
				return
			}
		}
	}

	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

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

type Task struct {
	ID    int
	Value string
}

type Result struct {
	TaskID int
	Output string
}

func worker(id int, tasks <-chan Task, results chan<- Result) {
	for t := range tasks {
		// Simulate work
		output := fmt.Sprintf("Worker %d processed Task %d: %s", id, t.ID, t.Value)
		results <- Result{TaskID: t.ID, Output: output}
	}
}

func main() {
	numWorkers := 3
	numTasks := 10

	tasks := make(chan Task, numTasks)
	results := make(chan Result, numTasks)

	// Start workers
	for w := 1; w <= numWorkers; w++ {
		go worker(w, tasks, results)
	}

	// Send tasks
	for j := 1; j <= numTasks; j++ {
		tasks <- Task{ID: j, Value: fmt.Sprintf("Data %d", j)}
	}
	close(tasks)

	// Collect results
	for a := 1; a <= numTasks; a++ {
		<-results
	}
}

func fetchResource(ctx context.Context, url string) (string, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return "", err
	}

	client := http.DefaultClient
	resp, err := client.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	// process response...
	return "Success", nil
}

func main() {
	concurrencyLimit := 5
	semaphore := make(chan struct{}, concurrencyLimit)

	var wg sync.WaitGroup
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			
			// Acquire semaphore
			semaphore <- struct{}{}
			
			// Critical section
			fmt.Printf("Goroutine %d executing\n", id)
			time.Sleep(time.Second) // Simulate work
			
			// Release semaphore
			<-semaphore
		}(i)
	}
	wg.Wait()
}

func orDone(ctx context.Context, c <-chan interface{}) <-chan interface{} {
	valStream := make(chan interface{})
	go func() {
		defer close(valStream)
		for {
			select {
			case <-ctx.Done():
				return
			case v, ok := <-c:
				if !ok {
					return
				}
				select {
				case valStream <- v:
				case <-ctx.Done():
				}
			}
		}
	}()
	return valStream
}

func tee(ctx context.Context, in <-chan interface{}) (<-chan interface{}, <-chan interface{}) {
	out1 := make(chan interface{})
	out2 := make(chan interface{})
	
	go func() {
		defer close(out1)
		defer close(out2)
		
		for val := range orDone(ctx, in) {
			// Create local copies of channels to track which ones we've sent to
			var out1Local, out2Local = out1, out2
			
			for i := 0; i < 2; i++ {
				select {
				case <-ctx.Done():
					return
				case out1Local <- val:
					out1Local = nil // Prevent sending to this channel again
				case out2Local <- val:
					out2Local = nil // Prevent sending to this channel again
				}
			}
		}
	}()
	
	return out1, out2
}

package main

import (
	"context"
	"fmt"
	"golang.org/x/sync/errgroup"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	
	g, ctx := errgroup.WithContext(ctx)
	
	urls := []string{
		"http://example.com",
		"http://example.net",
		"http://example.org",
	}
	
	for _, url := range urls {
		// Launch a goroutine to fetch the URL.
		url := url // https://golang.org/doc/faq#closures_and_goroutines
		g.Go(func() error {
			// Use the context returned by errgroup.WithContext.
			// If any goroutine returns an error, this context will be canceled.
			// return fetchResource(ctx, url)
			fmt.Printf("Fetching %s\n", url)
			return nil
		})
	}
	
	// Wait for all HTTP fetches to complete.
	if err := g.Wait(); err != nil {
		fmt.Printf("Failed to fetch all URLs: %v\n", err)
	} else {
		fmt.Println("Successfully fetched all URLs.")
	}
}