当取消传入请求时，取消 python 服务器中的下游 gRPC 请求

Question

当同步 Python gRPC 服务器的传入请求被取消时，是否有一个简单的模式可以遵循来取消所有工作和所有下游 gRPC 请求？

我认为这已被处理，但此页面表明它不是： https: //github.com/grpc/grpc/tree/master/examples/python/cancellation

请务必记住，gRPC Python 服务器由固定大小的线程池支持。当 RPC 被取消时，库不会终止您的服务线程。作为应用程序作者，您有责任确保您的服务线程在 RPC 取消后立即终止。

他们通过传递一个 threading.Event() 来处理它，这对于深度调用堆栈来说似乎很麻烦。还有其他选择吗？

Answer 1

我是您引用的文档的作者。

恐怕除了应用程序代码手动检测停止服务线程的机制之外，确实没有任何其他选择。这是因为（在 Posix 或 Windows 中）没有“杀死线程”的机制。

如果您担心的是 threading.Event 的实际管道，而不是事件本身，您可以使用 contextvars 来访问Event 以线程安全的方式，看起来大致类似于全局变量的使用。

Answer 2

上面的答案确实是正确的，但仍然有一种模式可以让您温和且可管理地终止线程。

为了向线程传输有关完成的信息，有一种编程模式——取消令牌。它在其他编程语言中很流行，例如 C# 和 Go （但这里的调用方式不同）。到目前为止，Python 中还没有这种模式的合理实现，但现在我已经编写了它。

安装库 cantok：

pip install cantok

并像示例中的代码一样使用：

from random import randint
from threading import Thread

from cantok import ConditionToken, CounterToken, TimeoutToken


counter = 0

def function(token):
    global counter
    while not token.cancelled:
        counter += 1

token = ConditionToken(lambda: randint(1, 100_000) == 1984) + CounterToken(400_000, direct=False) + TimeoutToken(1)
thread = Thread(target=function, args=(token, ))
thread.start()
thread.join()

print(counter)

我将简要解释此代码示例中发生的情况。我们在单独的线程中运行一个函数，并将取消标记传递给该函数。该函数会递增计数器，直到取消标记的 cancelled 属性变为 True。当以下事件之一发生时，此属性变为 True：令牌被取消、1 秒的时间限制结束、发生不太可能的事件（概率为十亿分之一的随机数等于1984）或者周期经过了 40 万次迭代。一旦发生这种情况，我们就退出循环并打印计数器值。

正如您所看到的，此模式允许您轻松管理对代码持续时间的一组复杂限制，并根据需要轻松取消操作。在文档中了解有关该库功能的更多信息。