当新请求来自同一会话中的同一用户时如何取消先前的请求答案

【问题标题】：how to cancel a previous requests when new request come from the same user in the same session当新请求来自同一会话中的同一用户时如何取消先前的请求
【发布时间】：2017-05-29 06:32:45
【问题描述】：

我们正在使用 aiohttp 构建一个 rest api。我们的应用程序旨在让用户发送请求比接收响应更频繁（因为计算时间）。对于用户而言，仅最新请求的结果很重要。是否可以停止对过期请求的计算？

谢谢

【问题讨论】：

这违反了 HTTP 的核心原则之一。如果您想将该服务扩展到负载均衡器后面的多个服务器怎么办？然后它会变得非常棘手，取消其他服务器可能正在处理的请求。这可能不是您应该尝试解决的实际问题！
“取消”到底是什么意思？服务器断开连接？
重写后：这是否意味着一个 HTTP 请求可能需要几分钟 (?) 才能完成？这也是糟糕的 HTTP 设计；一个请求最多应在几秒钟内得到答复，典型的基线目标是 200 毫秒。如果您需要运行更长时间的作业，请将其视为作业：一个 HTTP 请求启动一个后台任务并返回任务 ID。然后，客户端可以使用另一个 HTTP 请求查询作业状态。例如。 POST /tasks { params }，GET /tasks/42。显然，此类后台任务可以通过任何在您的后端有意义的机制来取消。
我们现在的目标是 5-10 秒。
对于 HTTP 请求来说太长了。根据POST /tasks 接受一个新的“作业请求”，启动一个独立于 Web 服务器的后台任务，返回一个新的资源 ID。让客户端查询新资源的结果/状态 (GET /tasks/42)。如果有新任务取代任务，则取消任务；客户端可以知道这种隐式取消，不再需要查询旧资源。

标签： python http nginx python-asyncio aiohttp

【解决方案1】：

您正在构建一些非常不像 HTTP 的东西。 HTTP 请求的响应时间不应超过几毫秒，并且 HTTP 请求不应相互依赖；如果您需要执行耗时较长的计算，请尝试通过更改架构/模型/缓存/其他方式来加快计算速度，或者将其明确视为长时间运行的作业可以通过 HTTP 接口进行控制。这意味着“工作”是可以通过HTTP查询的“物理资源”。您通过 POST 请求创建资源：

POST /tasks
Content-Type: application/json

{"some": "parameters", "go": "here"}

{"resource": "/tasks/42"}

然后就可以查询任务的状态了：

GET /tasks/42

{"status": "pending"}

并最终得到结果：

GET /tasks/42

{"status": "done", "results": [...]}

当您发布一个取代旧任务的新任务时，您的后端可以以任何它认为合适的方式取消旧任务；然后，该资源将返回“已取消”或类似状态。您的客户端在开始新任务后根本不会再次查询旧资源。

即使您的客户端每秒查询一次资源，它仍然会在服务器上使用较少的资源（一个连接打开 10 秒，而 10 个连接在同一时间范围内打开 200 毫秒），特别是如果您申请一些智能缓存。这也更具可扩展性，因为您可以独立于 HTTP 前端扩展任务后端，并且 HTTP 前端可以轻松扩展到多个服务器和负载平衡器。

【讨论】：

【解决方案2】：

我将发布来自@Drizzt1991 的解决方案：

你好，阿尔乔姆。你有一个很奇怪的要求。了解如果 1 个客户端正在使用带有 keep-alive 的套接字，则在回答第一个请求之前，实际上不可能看到下一个请求。这就是 HTTP 的工作方式，它在发送另一个请求之前期望结果。因此，只有当客户端将在 2 个单独的套接字上运行时，您的情况才有效，但是您需要再次断言，来自同一客户端的 2 个套接字将在同一台机器上路由。通过实践，这对于故障转移和其他东西并没有那么好。基本上它将是一个有状态的 API。即使你做了所有这些，也不是所有的图书馆都支持取消。传统的关系数据库只会忽略结果，但仍会处理待处理的查询。如果你做图遍历这样复杂的事情，而且你有很多步骤，那没关系，你可以取消。

但是，如果您断言，该客户端使用套接字池，并且它们被路由到同一台机器并且请求从取消中受益，那么这样的事情应该可以解决问题：

import asyncio
import random

from aiohttp import web


def get_session_id(request):
    # I don't know how you do session management, so left it out
    return ""


async def handle(request):
    session_id = get_session_id(request)
    request['tr_id'] = tr_id = int(random.random() * 1000000)

    running_tasks = request.app['running_tasks']
    if session_id in running_tasks and not running_tasks[session_id].done():
        running_tasks[session_id].cancel()
        del running_tasks[session_id]

    current_task = asyncio.ensure_future(_handle_impl(request))
    running_tasks[session_id] = current_task

    try:
        resp = await current_task
    except asyncio.CancelledError:
        print("Cancelled request", tr_id)
        resp = web.Response(text="Cancelled {}".format(tr_id))
    finally:
        if running_tasks[session_id] is current_task:
            del running_tasks[session_id]
    return resp


async def _handle_impl(request):
    tr_id = request['tr_id']
    print("Start request", tr_id)
    await asyncio.sleep(10)
    print("Finished request", tr_id)
    return web.Response(text="Finished {}".format(tr_id))


app = web.Application()
app.router.add_get('/', handle)
app.router.add_get('/{name}', handle)

app['running_tasks'] = {}

web.run_app(app, host="127.0.0.1", port=8080)

【讨论】：