Go语言中基于Channel的并发注册表与任务管理模式优化

type JobRegistry struct {
  submission chan JobRegistrySubmitRequest
  listing    chan JobRegistryListRequest
}

type JobRegistrySubmitRequest struct {
  request JobSubmissionRequest
  response chan Job
}

type JobRegistryListRequest struct {
  response chan []Job
}

// NewJobRegistry 和相关方法略

import (
    "errors"
    "fmt"
    "time"
)

// Job 示例结构体，假设已定义
type Job struct {
    Id   string
    Name string
    // ... 其他字段
}

// MakeJob 示例函数
func MakeJob(req JobSubmissionRequest) Job {
    // 实际业务逻辑创建Job
    return Job{Id: fmt.Sprintf("job-%d", time.Now().UnixNano()), Name: req.JobName}
}

// JobSubmissionRequest 示例结构体
type JobSubmissionRequest struct {
    JobName string
}

// OperationResult 用于封装操作结果和错误
type OperationResult struct {
    Value interface{} // 操作结果
    Err   error       // 操作错误
}

// RegistryOperation 定义所有注册表操作的通用接口
type RegistryOperation interface {
    // Execute 方法在JobManager的内部goroutine中执行，
    // 接收共享的jobMap作为参数，并执行具体操作。
    Execute(jobMap map[string]Job)
}

// SubmitOperation 提交任务操作
type SubmitOperation struct {
    Request  JobSubmissionRequest
    Response chan OperationResult // 用于返回提交结果
}

// Execute 实现RegistryOperation接口，处理任务提交逻辑
func (s *SubmitOperation) Execute(jobMap map[string]Job) {
    job := MakeJob(s.Request)
    jobMap[job.Id] = job
    s.Response <- OperationResult{Value: job, Err: nil} // 返回新创建的Job
}

// ListOperation 列出所有任务操作
type ListOperation struct {
    Response chan OperationResult // 用于返回任务列表
}

// Execute 实现RegistryOperation接口，处理任务列表逻辑
func (l *ListOperation) Execute(jobMap map[string]Job) {
    res := make([]Job, 0, len(jobMap))
    for _, v := range jobMap {
        res = append(res, v)
    }
    l.Response <- OperationResult{Value: res, Err: nil} // 返回Job列表
}

// JobManager 管理Job的注册表
type JobManager struct {
    operations chan RegistryOperation // 单一Channel接收所有操作
    quit       chan struct{}        // 用于优雅关闭
}

// NewJobManager 创建并启动JobManager
func NewJobManager() *JobManager {
    mgr := &JobManager{
        operations: make(chan RegistryOperation, 100), // 带有缓冲，防止阻塞
        quit:       make(chan struct{}),
    }

    go func() {
        jobMap := make(map[string]Job) // 共享状态，仅在此goroutine中访问
        for {
            select {
            case op := <-mgr.operations:
                op.Execute(jobMap) // 执行操作
            case <-mgr.quit:
                // 收到退出信号，关闭所有操作Channel并退出
                close(mgr.operations)
                return
            }
        }
    }()
    return mgr
}

// Close 用于优雅关闭JobManager
func (jm *JobManager) Close() {
    close(jm.quit)
}

// Submit 提交任务的客户端接口
func (jm *JobManager) Submit(req JobSubmissionRequest) (Job, error) {
    resChan := make(chan OperationResult, 1) // 创建一个临时的响应Channel
    op := &SubmitOperation{Request: req, Response: resChan}

    select {
    case jm.operations <- op: // 将操作发送到JobManager
        // 等待结果或超时
        select {
        case result := <-resChan:
            if result.Err != nil {
                return Job{}, result.Err
            }
            // 类型断言，确保返回正确的Job类型
            if job, ok := result.Value.(Job); ok {
                return job, nil
            }
            return Job{}, errors.New("unexpected result type from submit operation")
        case <-time.After(5 * time.Second): // 设置一个超时
            return Job{}, errors.New("submit operation timed out")
        }
    case <-time.After(1 * time.Second): // 如果发送操作本身也可能阻塞
        return Job{}, errors.New("failed to send submit operation to manager")
    }
}

// List 列出所有任务的客户端接口
func (jm *JobManager) List() ([]Job, error) {
    resChan := make(chan OperationResult, 1)
    op := &ListOperation{Response: resChan}

    select {
    case jm.operations <- op:
        select {
        case result := <-resChan:
            if result.Err != nil {
                return nil, result.Err
            }
            if jobs, ok := result.Value.([]Job); ok {
                return jobs, nil
            }
            return nil, errors.New("unexpected result type from list operation")
        case <-time.After(5 * time.Second):
            return nil, errors.New("list operation timed out")
        }
    case <-time.After(1 * time.Second):
        return nil, errors.New("failed to send list operation to manager")
    }
}