0

我在 python 中使用 falcon 框架来形成 web api 的 json 响应。

例如，我有一个名为的函数logic()，它可以工作 30-90 分钟。我想要这样的东西：

  1. 当 http-client 请求 /api/somepath.json 我们调用 somepath_handle()
  2. somepath_handle()logic()在另一个线程/进程中运行
  3. 完成logic()后，线程关闭
  4. somepath_handle()logic()从返回中读取响应
  5. 如果在完成somepath_handle()之前被杀死logic()，那么线程/等在完成之前logic()不会停止

编码：

def somepath_handle():
    run_async_logic()
    response=wait_for_async_logic_response() # read response of logic()
    return_response(response)
4

2 回答 2

1

如果您的过程需要这么长时间，我建议您使用电子邮件或实时通知系统将结果发送给用户？

于 2015-07-15T09:02:20.653 回答
0

我正在使用一个简单的工作者来创建我正在处理一些命令的队列。如果添加简单的响应存储，则有可能处理任何请求并且在连接丢失时不会丢失它们。

示例：它是使用 falconframework.org 响应请求的主要功能。

主要.py：

from flow import Flow
import falcon
import threading
import storage

__version__ = 0.1
__author__ = 'weldpua2008@gmail.com'


app = falcon.API(
    media_type='application/json')

app.add_route('/flow', Flow())

THREADS_COUNT = 1
# adding the workers to process queue of command
worker = storage.worker
for _ in xrange(THREADS_COUNT):
    thread = threading.Thread(target=worker)
    thread.daemon = True
    thread.start()

这是带有工作代码 storage.py 的简单存储：

from Queue import Queue
import subprocess
import logging
main_queque = Queue()

def worker():
    global main_roles_queque
    while True:
        try:
            cmd = main_queque.get()
            #do_work(item)
            #time.sleep(5)
            handler = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
            stdout, stderr = handler.communicate()
            logging.critical("[queue_worker]: stdout:%s, stderr:%s, cmd:%s" %(stdout, stderr, cmd))
            main_queque.task_done()
        except Exception as error:
            logging.critical("[queue_worker:error] %s" %(error))

它将处理任何请求 [POST, GET] flow.py 的类：

import storage
import json
import falcon
import random

class Flow(object):

    def on_get(self, req, resp):
        storage_value = storage.main_queque.qsize()
        msg = {"qsize": storage_value}
        resp.body = json.dumps(msg, sort_keys=True, indent=4)
        resp.status = falcon.HTTP_200

    #curl -H "Content-Type: application/json" -d '{}'  http://10.206.102.81:8888/flow
    def on_post(self, req, resp):
        r = random.randint(1, 10000000000000)
        cmd = 'sleep 1;echo "ss %s"' % str(r)
        storage.main_queque.put(cmd)
        storage_value = cmd
        msg = {"value": storage_value}
        resp.body = json.dumps(msg, sort_keys=True, indent=4)
        resp.status = falcon.HTTP_200
于 2015-09-09T04:34:08.250 回答