Go 并发模式：使用带回复的 Registry 模式实现安全的数据访问

package main

import (
    "fmt"
    "io"
    "os"
    "strconv"
    "sync"
)

// Job 接口定义了 Job 的基本行为
type Job interface {
    Run()
    Serialize(io.Writer)
    GetID() string
}

// IntJob 是一个具体的 Job 实现
type IntJob struct {
    ID   string
    Data int
    out  chan int
}

func (job *IntJob) GetID() string {
    return job.ID
}

// GetOutChan 返回用于接收结果的 channel
func (job *IntJob) GetOutChan() chan int {
    return job.out
}

// Run 执行 Job 的具体逻辑
func (job *IntJob) Run() {
    // 模拟耗时操作
    result := job.Data * 2
    job.out <- result
    close(job.out)
}

// Serialize 将 Job 序列化到 io.Writer
func (job *IntJob) Serialize(o io.Writer) {
    _, err := o.Write([]byte(fmt.Sprintf("IntJob: ID=%s, Data=%d", job.ID, job.Data)))
    if err != nil {
        fmt.Println("序列化失败:", err)
    }
}

// JobRegistry 定义了 Registry 的结构
type JobRegistry struct {
    submission chan JobRegistrySubmitRequest
    listing    chan JobRegistryListRequest
    mu         sync.Mutex // 使用互斥锁保护 jobMap
}

// JobRegistrySubmitRequest 定义了提交 Job 的请求
type JobRegistrySubmitRequest struct {
    request  Job
    response chan string // 返回 Job ID
}

// JobRegistryListRequest 定义了列出所有 Job 的请求
type JobRegistryListRequest struct {
    response chan []Job
}

// NewJobRegistry 创建一个新的 JobRegistry
func NewJobRegistry() *JobRegistry {
    registry := &JobRegistry{
        submission: make(chan JobRegistrySubmitRequest, 10),
        listing:    make(chan JobRegistryListRequest, 10),
        mu:         sync.Mutex{},
    }

    go registry.run()

    return registry
}

// run 是 Registry 的主循环，负责处理请求
func (this *JobRegistry) run() {
    jobMap := make(map[string]Job)

    for {
        select {
        case sub := <-this.submission:
            job := sub.request
            this.mu.Lock()
            jobMap[job.GetID()] = job
            this.mu.Unlock()
            sub.response <- job.GetID()
            close(sub.response)

            go func(j Job) {
                j.Run()
            }(job)

        case list := <-this.listing:
            res := make([]Job, 0, 100)
            this.mu.Lock()
            for _, v := range jobMap {
                res = append(res, v)
            }
            this.mu.Unlock()
            list.response <- res
            close(list.response)
        }
    }
}

// Submit 提交一个 Job 到 Registry
func (this *JobRegistry) Submit(job Job) (string, error) {
    res := make(chan string, 1)
    req := JobRegistrySubmitRequest{request: job, response: res}
    this.submission <- req
    jobID := <-res
    return jobID, nil
}

// List 列出 Registry 中的所有 Job
func (this *JobRegistry) List() ([]Job, error) {
    res := make(chan []Job, 1)
    req := JobRegistryListRequest{response: res}
    this.listing <- req
    jobs := <-res
    return jobs, nil
}

func main() {
    registry := NewJobRegistry()

    // 提交 Job
    job1 := &IntJob{ID: "job1", Data: 10, out: make(chan int, 1)}
    job2 := &IntJob{ID: "job2", Data: 20, out: make(chan int, 1)}

    id1, _ := registry.Submit(job1)
    id2, _ := registry.Submit(job2)

    fmt.Println("提交的 Job ID:", id1, id2)

    // 列出所有 Job
    jobs, _ := registry.List()
    fmt.Println("Registry 中的 Job 数量:", len(jobs))

    // 获取 Job 结果
    result1 := <-job1.GetOutChan()
    result2 := <-job2.GetOutChan()

    fmt.Println("Job 1 的结果:", result1)
    fmt.Println("Job 2 的结果:", result2)

    // 序列化 Job
    file, err := os.Create("job1.txt")
    if err != nil {
        fmt.Println("创建文件失败:", err)
        return
    }
    defer file.Close()
    job1.Serialize(file)
}