怎样用Golang构建高效流水线模式 解析channel的缓冲与关闭策略

package main

import (
    "fmt"
    "image"
    "image/jpeg"
    "image/png"
    "io"
    "log"
    "os"
    "path/filepath"
    "strconv"
    "sync"
    "time"

    "github.com/nfnt/resize"
)

// ImageTask represents a single image processing task.
type ImageTask struct {
    InputPath  string
    OutputPath string
    Width      uint
    Height     uint
}

// resizeImage resizes the image and returns the resized image.
func resizeImage(img image.Image, width, height uint) image.Image {
    return resize.Resize(width, height, img, resize.Lanczos3)
}

// decodeImage decodes the image from the given reader.
func decodeImage(reader io.Reader, inputPath string) (image.Image, string, error) {
    ext := filepath.Ext(inputPath)
    switch ext {
    case ".jpg", ".jpeg":
        img, err := jpeg.Decode(reader)
        if err != nil {
            return nil, "", fmt.Errorf("decoding JPEG: %w", err)
        }
        return img, ".jpg", nil
    case ".png":
        img, err := png.Decode(reader)
        if err != nil {
            return nil, "", fmt.Errorf("decoding PNG: %w", err)
        }
        return img, ".png", nil
    default:
        return nil, "", fmt.Errorf("unsupported image format: %s", ext)
    }
}

// worker reads image tasks from the tasks channel, processes them, and sends the results to the results channel.
func worker(id int, tasks <-chan ImageTask, results chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range tasks {
        startTime := time.Now()

        // Open the input file.
        inputFile, err := os.Open(task.InputPath)
        if err != nil {
            log.Printf("Worker %d: Error opening input file %s: %v", id, task.InputPath, err)
            results <- fmt.Sprintf("Error processing %s: %v", task.InputPath, err)
            continue
        }
        defer inputFile.Close()

        // Decode the image.
        img, ext, err := decodeImage(inputFile, task.InputPath)
        if err != nil {
            log.Printf("Worker %d: Error decoding image %s: %v", id, task.InputPath, err)
            results <- fmt.Sprintf("Error processing %s: %v", task.InputPath, err)
            continue
        }

        // Resize the image.
        resizedImage := resizeImage(img, task.Width, task.Height)

        // Create the output file.
        outputFile, err := os.Create(task.OutputPath)
        if err != nil {
            log.Printf("Worker %d: Error creating output file %s: %v", id, task.OutputPath, err)
            results <- fmt.Sprintf("Error processing %s: %v", task.InputPath, err)
            continue
        }
        defer outputFile.Close()

        // Encode and save the resized image.
        switch ext {
        case ".jpg", ".jpeg":
            err = jpeg.Encode(outputFile, resizedImage, nil)
        case ".png":
            err = png.Encode(outputFile, resizedImage)
        }

        if err != nil {
            log.Printf("Worker %d: Error encoding image %s: %v", id, task.OutputPath, err)
            results <- fmt.Sprintf("Error processing %s: %v", task.InputPath, err)
            continue
        }

        duration := time.Since(startTime)
        results <- fmt.Sprintf("Worker %d: Successfully processed %s in %v", id, task.InputPath, duration)
    }
}

func main() {
    // Configuration
    numWorkers := 4
    inputDir := "input_images"
    outputDir := "output_images"
    targetWidth := uint(800)
    targetHeight := uint(600)

    // Create input and output directories if they don't exist.
    if _, err := os.Stat(inputDir); os.IsNotExist(err) {
        log.Fatalf("Input directory '%s' does not exist. Please create it and add images.", inputDir)
    }
    if _, err := os.Stat(outputDir); os.IsNotExist(err) {
        err := os.MkdirAll(outputDir, 0755)
        if err != nil {
            log.Fatalf("Failed to create output directory: %v", err)
        }
    }

    // Create channels for tasks and results.
    tasks := make(chan ImageTask, 100) // Buffered channel
    results := make(chan string, 100) // Buffered channel

    // Start the workers.
    var wg sync.WaitGroup
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(i, tasks, results, &wg)
    }

    // Read image files from the input directory and create tasks.
    filepath.Walk(inputDir, func(path string, info os.FileInfo, err error) error {
        if err != nil {
            return err
        }
        if info.IsDir() {
            return nil
        }
        outputPath := filepath.Join(outputDir, "resized_"+info.Name())
        tasks <- ImageTask{
            InputPath:  path,
            OutputPath: outputPath,
            Width:      targetWidth,
            Height:     targetHeight,
        }
        return nil
    })

    // Close the tasks channel after all tasks have been sent.
    close(tasks)

    // Wait for all workers to complete.
    go func() {
        wg.Wait()
        close(results) // Close the results channel after all workers are done.
    }()

    // Collect and print the results.
    for result := range results {
        fmt.Println(result)
    }

    fmt.Println("Image processing completed.")
}

// 生产者
func producer(ch chan int) {
    defer close(ch) // 确保在函数退出时关闭channel
    for i := 0; i < 10; i++ {
        ch <- i
    }
}

// 消费者
func consumer(ch chan int) {
    for val := range ch { // 使用range循环遍历channel，channel关闭时循环自动结束
        fmt.Println(val)
    }
}

func main() {
    ch := make(chan int, 5)
    go producer(ch)
    consumer(ch)
}

type Result struct {
    Data int
    Err  error
}

func worker(input <-chan int, output chan<- Result) {
    for num := range input {
        // 模拟可能发生的错误
        if num%2 == 0 {
            output <- Result{Data: num * 2, Err: nil}
        } else {
            output <- Result{Data: 0, Err: fmt.Errorf("invalid number: %d", num)}
        }
    }
}

func main() {
    input := make(chan int, 10)
    output := make(chan Result, 10)

    go worker(input, output)

    for i := 0; i < 10; i++ {
        input <- i
    }
    close(input)

    for i := 0; i < 10; i++ {
        result := <-output
        if result.Err != nil {
            fmt.Println("Error:", result.Err)
        } else {
            fmt.Println("Result:", result.Data)
        }
    }
    close(output)
}

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

func worker(ctx context.Context, id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case job, ok := <-jobs:
            if !ok {
                fmt.Printf("Worker %d: jobs channel closed\n", id)
                return
            }
            fmt.Printf("Worker %d: processing job %d\n", id, job)
            time.Sleep(time.Second) // Simulate work
            results <- job * 2
        case <-ctx.Done():
            fmt.Printf("Worker %d: received shutdown signal\n", id)
            return
        }
    }
}

func main() {
    numWorkers := 3
    numJobs := 5

    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel() // Ensure cancellation signal is sent when main exits

    var wg sync.WaitGroup
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(ctx, i, jobs, results, &wg)
    }

    // Send jobs
    for i := 1; i <= numJobs; i++ {
        jobs <- i
    }
    close(jobs) // Signal no more jobs

    // Collect results (or handle them concurrently)
    go func() {
        wg.Wait()           // Wait for all workers to finish
        close(results) // Close the results channel after all workers are done.
    }()

    // Simulate a shutdown signal after some time
    time.Sleep(3 * time.Second)
    fmt.Println("Sending shutdown signal...")
    cancel() // Signal all workers to stop

    // Print results
    for result := range results {
        fmt.Println("Result:", result)
    }

    fmt.Println("Program finished")
}