如何在golang中根据ID取消通道作业

在golang中，取消通道作业是一个常见的需求。当我们在处理并发任务时，有时候需要根据任务的ID来取消正在执行的任务。那么，在golang中如何实现这个功能呢？下面，我将为大家介绍一种简单有效的方法。首先，我们需要创建一个带缓冲的通道，用于存储任务的ID。然后，我们可以使用select语句来监听通道的读取操作，并在需要取消任务时，将任务的ID发送到通道中。接着，在任务的执行函数中，我们可以通过判断通道是否被关闭来确定是否需要取消任务。如果通道被关闭，说明任务已被取消，我们可以在适当的地方终止任务的执行。通过这种方式，我们可以方便地在golang中根据ID取消通道作业。以上就是php小编新一为大家介绍的方法，希望对大家有所帮助！

问题内容

所以我有一个 post 端点，它创建一个作业并将它们添加到一个 chan 中。 workerjobschan = make(chan job, maxqueuesize)

这就是我在频道中执行作业的方式（main.go）：

for i := 1; i <= maxworkers; i++ {
    go func(i int) {
        for job := range workerjobschan {
            ctx, cancel := context.withcancel(context.background())
            storejob(job.search.id, cancel)
            job.execute(ctx, c.db, i)
        }
    }(i)
}

我将取消函数存储在地图中：canceljobfuncs = make(map[int]context.cancelfunc)。

这是工作职能：

func (j *job) execute(ctx context.context, db *sql.db, workerid int) error {


    for {
        select {
        // check for cancellation signal
        case <-ctx.done():
             if err := ctx.err(); err != nil {
                fmt.println("worker", workerid, "error", err)
             }
            fmt.println("worker", workerid, "cancelled")
            return nil

        default:
            fmt.printf("worker%d: processing %s\n", workerid, j.search.query)
            time.sleep(2 * time.second)
            fmt.printf("worker%d: active %s\n", workerid, j.search.query)
            time.sleep(5 * time.second)
            fmt.printf("worker%d: completed %s!\n", workerid, j.search.query)

        }
    }
}

我取消这样的上下文（在http处理程序中）：

cancelJob(search.ID)

但该作业仍在继续运行。我尝试了很多东西，但似乎无法让它发挥作用。

解决方法

这里有一种方法可以说明一个明显的观点：如果您的代码不检查 ctx.done()，它就无法知道它已被取消。
（顺便说一句，这是@jimb 在对您的问题的评论中所写内容的另一种改写）。

因此，当 .execute(...) 方法中的代码开始执行此块时：

fmt.printf("worker%d: processing %s\n", workerid, j.search.query)
    time.sleep(2 * time.second)
    fmt.printf("worker%d: active %s\n", workerid, j.search.query)
    time.sleep(5 * time.second)
    fmt.printf("worker%d: completed %s!\n", workerid, j.search.query)

它将到达该块的末尾（7 秒）。没有任何指令告诉它在取消时停止。

如果您希望函数能够在“睡眠”指令期间检测取消，则必须更改代码。

以下是如何使用您的示例执行此操作的一个示例：

func (j *job) execute(ctx context.context, db *sql.db, workerid int) error {


    for {
        fmt.printf("worker%d: processing %s\n", workerid, j.search.query)

        // rewrite time.sleep() with time.after() so that it can be composed
        // in a select statement:
        select {
        case <-ctx.done():
            fmt.println("worker", workerid, "cancelled")
            return nil
        case <-time.after(2 * time.second):
            // keep going
        }

        fmt.printf("worker%d: active %s\n", workerid, j.search.query)

        select {
        case <-ctx.done():
            fmt.println("worker", workerid, "cancelled")
            return nil
        case <-time.after(5 * time.second):
            // keep going
        }

        fmt.printf("worker%d: completed %s!\n", workerid, j.search.query)
    }
}

https://www.php.cn/link/3bc31a430954d8326605fc690ed22f4d

我猜你的实际代码没有 time.sleep() 指令，而是 processsearch(...) 或 doquery(...) 或 ...

如果您需要这些函数在执行期间可取消，则需要以某种方式将取消上下文传递给它们，并让它们以某种方式检查取消情况。

“传递上下文”的一种方法显然是将其添加到所述函数的参数中：

processsearch(ctx, ...)
doquery(ctx, ...)

但根据您现有的代码，某些参数可能已经有内置的方法可以取消。
举例来说：

// an http.Request carries a context:
func doQuery(req *http.Request, ....) {
    ...
}

// at call site:
    ...
    req := http.NewRequestWithContext(ctx, "GET", "https://some.other.service/", nil)
    doQuery(req, ...)

以上是如何在golang中根据ID取消通道作业的详细内容。更多信息请关注PHP中文网其他相关文章！