Goroutine-停止运行进程[重复]

Question

我使用以下代码在大多数情况下都可以正常工作，以防我们使用一些长时间运行的进程，它不会在程序内停止不结束（这里我限制为 60 秒例子）

我希望每个作业都在 5 秒后终止（即使它没有完成工作也终止进程）， 不 更改函数myLongRunningFunc，我该怎么做。

我知道这在 go 中解决起来并不简单，有什么我可以使用的技巧吗？

这是一些可重现的最小示例

https://play.golang.org/p/a0RWY4bYWMt

package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    "github.com/gammazero/workerpool"
)

func main() {
    // here define a timeout for 5 sec,
    // the task should be terminate after 5 sec
    ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
    defer cancel()

    runner := newRunner(ctx, 10)

    runner.do(job{
        Name: "a",
        Task: func() jobResult {
            select {
            case <-ctx.Done():
                return jobResult{Error: errors.New("Timedout, exiting")}
            default:
                myLongRunningFunc("A job")
            }
            return jobResult{Data: "from a"}
        },
    })

    runner.do(job{
        Name: "b",
        Task: func() jobResult {
            select {
            case <-ctx.Done():
                return jobResult{Error: errors.New("Timeouts, exiting")}
            default:
                myLongRunningFunc("B job")
            }

            return jobResult{Data: "from b"}
        },
    })

    results := runner.getjobResults()
    fmt.Println(results)
    time.Sleep(time.Second * 60)
}

func myLongRunningFunc(name string) {
    for i := 0; i < 100000; i++ {
        time.Sleep(time.Second * 1)
        msg := "job" + name + " running..\n"
        fmt.Println(msg)
    }
}

type runner struct {
    *workerpool.WorkerPool
    ctx     context.Context
    kill    chan struct{}
    result  chan jobResult
    results []jobResult
}

func (r *runner) processResults() {
    for {
        select {
        case res, ok := <-r.result:
            if !ok {
                goto Done
            }
            r.results = append(r.results, res)
        }
    }
Done:
    <-r.kill
}

func newRunner(ctx context.Context, numRunners int) *runner {
    r := &runner{
        WorkerPool: workerpool.New(numRunners),
        ctx:        ctx,
        kill:       make(chan struct{}),
        result:     make(chan jobResult),
    }
    go r.processResults()
    return r
}

func (r *runner) do(j job) {
    r.Submit(r.wrap(&j))
}

func (r *runner) getjobResults() []jobResult {
    r.StopWait()
    close(r.result)
    r.kill <- struct{}{}
    return r.results
}

func (r *runner) wrap(job *job) func() {
    return func() {
        job.result = make(chan jobResult)
        go job.Run()
        select {
        case res := <-job.result:
            r.result <- res
        case <-r.ctx.Done():
            fmt.Printf("Job '%s' should stop here\n", job.Name)
            r.result <- jobResult{name: job.Name, Error: r.ctx.Err()}
        }
    }
}

type job struct {
    Name    string
    Task    func() jobResult
    Context context.Context
    result  chan jobResult
    stopped chan struct{}
    done    context.CancelFunc
}

func (j *job) Run() {
    result := j.Task()
    result.name = j.Name
    j.result <- result
}

type jobResult struct {
    name  string
    Error error
    Data  interface{}
}

由于我使用的是圣坛频道，因此编辑不相关

Answer 1

我希望每个作业都将被终止（即使它是终止进程 没有完成工作）5秒后，我怎么能不改变 函数 myLongRunningFunc。

然后你只需添加一个 5 秒的服务员然后退出。

package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    "github.com/gammazero/workerpool"
)

func main() {
    go func() {
        // here define a timeout for 5 sec,
        // the task should be terminate after 5 sec
        ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
        defer cancel()

        runner := newRunner(ctx, 10)

        runner.do(job{
            Name: "a",
            Task: func() jobResult {
                select {
                case <-ctx.Done():
                    return jobResult{Error: errors.New("Timedout, exiting")}
                default:
                    myLongRunningFunc("A job")
                }
                return jobResult{Data: "from a"}
            },
        })

        runner.do(job{
            Name: "b",
            Task: func() jobResult {
                select {
                case <-ctx.Done():
                    return jobResult{Error: errors.New("Timeouts, exiting")}
                default:
                    myLongRunningFunc("B job")
                }

                return jobResult{Data: "from b"}
            },
        })

        results := runner.getjobResults()
        fmt.Println(results)
        time.Sleep(time.Second * 60)
    }()
    <-time.After(time.Second * 5)
}

func myLongRunningFunc(name string) {
    for i := 0; i < 100000; i++ {
        time.Sleep(time.Second * 1)
        msg := "job" + name + " running..\n"
        fmt.Println(msg)
    }
}

type runner struct {
    *workerpool.WorkerPool
    ctx     context.Context
    kill    chan struct{}
    result  chan jobResult
    results []jobResult
}

func (r *runner) processResults() {
    for {
        select {
        case res, ok := <-r.result:
            if !ok {
                goto Done
            }
            r.results = append(r.results, res)
        }
    }
Done:
    <-r.kill
}

func newRunner(ctx context.Context, numRunners int) *runner {
    r := &runner{
        WorkerPool: workerpool.New(numRunners),
        ctx:        ctx,
        kill:       make(chan struct{}),
        result:     make(chan jobResult),
    }
    go r.processResults()
    return r
}

func (r *runner) do(j job) {
    r.Submit(r.wrap(&j))
}

func (r *runner) getjobResults() []jobResult {
    r.StopWait()
    close(r.result)
    r.kill <- struct{}{}
    return r.results
}

func (r *runner) wrap(job *job) func() {
    return func() {
        job.result = make(chan jobResult)
        go job.Run()
        select {
        case res := <-job.result:
            r.result <- res
        case <-r.ctx.Done():
            fmt.Printf("Job '%s' should stop here\n", job.Name)
            r.result <- jobResult{name: job.Name, Error: r.ctx.Err()}
        }
    }
}

type job struct {
    Name    string
    Task    func() jobResult
    Context context.Context
    result  chan jobResult
    stopped chan struct{}
    done    context.CancelFunc
}

func (j *job) Run() {
    result := j.Task()
    result.name = j.Name
    j.result <- result
}

type jobResult struct {
    name  string
    Error error
    Data  interface{}
}

Answer 2

我认为不可能从外部 goroutine 中停止 goroutine。你可以检查它是否超时，但是你不能停止它。

你可以做的是通过通道向goroutine发送一条消息，在这种情况下可以监控并停止。

You can find examples here