Go에서 강력한 작업 실행 컨텍스트 구축-Golang-php.cn

Go에서 강력한 작업 실행 컨텍스트 구축

Susan Sarandon

풀어 주다： 2025-01-01 01:02:10

원래의

275명이 탐색했습니다.

Building a Robust Task Execution Context in Go

이것이 Go에서의 오류 처리에 대한 나의 마지막 견해일 것입니다. 저도 이게 제일 좋은 것 같아요. 우리는 실행하는 모든 명령이 컨텍스트에 있음을 알고 있습니다. 그리고 컨텍스트에 오류가 있을 수 있습니다. 이것은 단순히 현재 컨텍스트 위에 래퍼를 만드는 것이 어떨까 생각한 때입니다. 따라서 모든 작업이 특정 fn을 통해 실행되면 ctx에 오류가 있는지 확인할 수 있으며, 그렇다면 실행하지 말고 오류를 실행하고 수집하세요. 이는 안티 패턴이 될 수도 있지만 그렇게 될 때까지 우리는 놀아볼 수 있습니다.

글쎄, 커서에 추가할 것이 거의 없었습니다 ->

문제

동시 작업을 처리할 때 다음과 같은 일반적인 문제를 고려하십시오.

여러 고루틴에서 오류 수집
스레드 안전성 유지
동시 실행 제한
모든 오류를 수집하면서 첫 번째 오류를 보존
오류 처리 패턴 정리

해결책: TaskContext

이러한 문제를 해결하는 TaskContext를 구축해 보겠습니다.

package taskctx

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

type RunFn[T any] func() (T, error)

type TaskContext struct {
    context.Context
    mu       sync.RWMutex
    err      error
    multiErr []error
}

func NewTaskContext(parent context.Context) *TaskContext {
    if parent == nil {
        panic("cannot create context from nil parent")
    }
    return &TaskContext{Context: parent}
}

로그인 후 복사

주요 특징

1. 스레드로부터 안전한 오류 처리

func (c *TaskContext) WithError(err error) *TaskContext {
    if err == nil {
        return c
    }

    c.mu.Lock()
    defer c.mu.Unlock()

    c.multiErr = append(c.multiErr, err)
    if c.err == nil {
        c.err = err
    } else {
        c.err = errors.Join(c.err, err)
    }
    return c
}

로그인 후 복사

2. 단일 작업 실행

func Run[T any](ctx *TaskContext, fn RunFn[T]) T {
    var zero T
    if err := ctx.Err(); err != nil {
        return zero
    }

    result, err := fn()
    if err != nil {
        ctx.WithError(err)
        return zero
    }
    return result
}

로그인 후 복사

3. 병렬 작업 실행

func RunParallel[T any](ctx *TaskContext, fns ...func() (T, error)) ([]T, error) {
    if err := ctx.Err(); err != nil {
        return nil, err
    }

    results := make([]T, len(fns))
    var resultsMu sync.Mutex
    var wg sync.WaitGroup
    wg.Add(len(fns))

    for i, fn := range fns {
        i, fn := i, fn
        go func() {
            defer wg.Done()
            result, err := fn()
            if err != nil {
                ctx.AddError(fmt.Errorf("task %d: %w", i+1, err))
            } else {
                resultsMu.Lock()
                results[i] = result
                resultsMu.Unlock()
            }
        }()
    }

    wg.Wait()
    return results, ctx.Errors()
}

로그인 후 복사

4. 동시성 제어

func RunParallelWithLimit[T any](ctx *TaskContext, limit int, fns ...func() (T, error)) ([]T, error) {
    // ... similar to RunParallel but with semaphore ...
    sem := make(chan struct{}, limit)
    // ... implementation ...
}

로그인 후 복사

사용 예

단순 작업 실행

func ExampleTaskContext_ShipmentProcessing() {
    ctx := goctx.NewTaskContext(context.Background())

    order := dummyOrder()
    shipment := dummyShipment()

    // Step 1: Validate address
    // Step 2: Calculate shipping cost
    // Step 3: Generate label
    _ = goctx.Run(ctx, validateAddress("123 Main St"))
    cost := goctx.Run(ctx, calculateShipping(order))
    trackingNum := goctx.Run(ctx, generateLabel(shipment.OrderID, cost))

    if ctx.Err() != nil {
        fmt.Printf("Error: %v\n", ctx.Err())
        return
    }

    shipment.Status = "READY"
    shipment.TrackingNum = trackingNum
    fmt.Printf("Shipment processed: %+v\n", shipment)

    // Output:
    // Shipment processed: {OrderID:ORD123 Status:READY TrackingNum:TRACK-ORD123-1234567890}
}

로그인 후 복사

병렬 작업 실행

func ExampleTaskContext_OrderProcessing() {
    ctx := goctx.NewTaskContext(context.Background())

    // Mock order
    order := []OrderItem{
        {ProductID: "LAPTOP", Quantity: 2},
        {ProductID: "MOUSE", Quantity: 3},
    }

    taskCtx := goctx.NewTaskContext(ctx)

    // Create inventory checks for each item
    inventoryChecks := goctx.Run[[]goctx.RunFn[bool]](taskCtx,
        func() ([]goctx.RunFn[bool], error) {
            return streams.NewTransformer[OrderItem, goctx.RunFn[bool]](order).
                Transform(streams.MapItSimple(checkInventory)).
                Result()
        })

    // Run inventory checks in parallel
    _, err := goctx.RunParallel(ctx, inventoryChecks...)
    fmt.Printf("Inventory check error: %v\n", err)

    // Output:
    // Inventory check error: task 1: insufficient inventory for LAPTOP
}

로그인 후 복사

이익

스레드 안전성: 모든 작업은 뮤텍스로 보호됩니다
오류 수집: 첫 번째 오류와 모든 오류를 모두 유지
컨텍스트 통합: Go의 컨텍스트 패키지와 작동
일반 지원: 모든 반환 유형에서 작동
동시성 제어: 병렬 실행 제한 지원 내장

테스트

구현을 테스트하는 방법은 다음과 같습니다.

func TestTaskContext(t *testing.T) {
    t.Run("handles parallel errors", func(t *testing.T) {
        ctx := NewTaskContext(context.Background())
        _, err := RunParallel(ctx,
            func() (int, error) { return 0, errors.New("error 1") },
            func() (int, error) { return 0, errors.New("error 2") },
        )
        assert.Error(t, err)
        assert.Contains(t, err.Error(), "error 1")
        assert.Contains(t, err.Error(), "error 2")
    })
}

로그인 후 복사