python – CherryPy：如何在数据更新时停止和缓冲传入的请求

第一个是基于插件的.

基于插件还需要一种同步.它只能起作用,因为只有一个BackgroundTask进行修改(也只是一个原子操作).

import time
import threading

import cherrypy
from cherrypy.process import plugins

UPDATE_INTERVAL = 0.5
REQUEST_DELAY = 0.1
UPDATE_DELAY = 0.1
THREAD_POOL_SIZE = 20

next_data = 1

class DataGateway(plugins.SimplePlugin):

    def __init__(self,bus):
        super(DataGateway,self).__init__(bus)
        self.data = next_data

    def start(self):
        self.bus.log("Starting DataGateway")
        self.bus.subscribe('dg:get',self._get_data)
        self.bus.subscribe('dg:update',self._update_data)
        self.bus.log("DataGateway has been started")

    def stop(self):
        self.bus.log("Stopping DataGateway")
        self.bus.unsubscribe('dg:get',self._get_data)
        self.bus.unsubscribe('dg:update',self._update_data)
        self.bus.log("DataGateway has been stopped")

    def _update_data(self,new_val):
        self.bus.log("Changing data,buffering should start!")
        self.data = new_val
        time.sleep(UPDATE_DELAY)
        self.bus.log("Continue serving buffered and new requests.")

    def _get_data(self):
        return self.data


def processData():
    """Backround task works for half hour three times a day,and when finishes it publish it in the engine buffer."""
    global next_data
    cherrypy.engine.publish("dg:update",next_data)
    next_data += 1


class DataPublisher(object):

    @property
    def data(self):
        return cherrypy.engine.publish('dg:get').pop()

    @cherrypy.expose
    def index(self):
        result = "I get " + str(self.data)
        cherrypy.engine.log(result)
        time.sleep(REQUEST_DELAY)
        return result

if __name__ == '__main__':
    conf = {
        'global': {
            'server.thread_pool': THREAD_POOL_SIZE,'server.socket_host': '127.0.0.1','server.socket_port': 8080,}
    }
    cherrypy.config.update(conf)
    DataGateway(cherrypy.engine).subscribe()
    plugins.BackgroundTask(UPDATE_DELAY,processData).start()
    cherrypy.quickstart(DataPublisher())

在这个版本中,同步来自于read& amp; write操作在cherrypy.engine线程上执行.所有内容都是在您刚刚操作发布到引擎中的插件DataGateway上抽象出来的.

第二种方法是使用Event a threading.Event对象.这是一种更加手动的方法,其额外的好处是,由于读取速度更快,因此它可能会更快,因为它不会在cherrypy.engine线程上执行.

threading.Event based(a.k.a.manual)

import time
import cherrypy
import threading
from cherrypy.process import plugins

UPDATE_INTERVAL = 0.5
REQUEST_DELAY = 0.1
UPDATE_DELAY = 0.1
THREAD_POOL_SIZE = 20

next_data = 1

def processData():
    """Backround task works for half hour three times a day,and when finishes it publish it in the engine buffer."""
    global next_data
    cherrypy.engine.publish("doChangeData",next_data)
    next_data += 1


class DataPublisher(object):

    def __init__(self):
        self._data = next_data
        self._data_readable = threading.Event()
        cherrypy.engine.subscribe('doChangeData',self.changeData)

    @property
    def data(self):
        if self._data_readable.is_set():
            return self._data
        else:
            self._data_readable.wait()
            return self.data

    @data.setter
    def data(self,value):
        self._data_readable.clear()
        time.sleep(UPDATE_DELAY)
        self._data = value
        self._data_readable.set()

    def changeData(self,buffering should start!")
        self.data = newData
        cherrypy.engine.log("Continue serving buffered and new requests.")

    @cherrypy.expose
    def index(self):
        result = "I get " + str(self.data)
        cherrypy.engine.log(result)
        time.sleep(REQUEST_DELAY)
        return result

if __name__ == '__main__':
    conf = {
        'global': {
            'server.thread_pool': THREAD_POOL_SIZE,}
    }
    cherrypy.config.update(conf)
    plugins.BackgroundTask(UPDATE_INTERVAL,processData).start()
    cherrypy.quickstart(DataPublisher())

我已经在@property装饰器中添加了一些细节,但真正的要点是在threading.Event以及DataPublisher对象在工作线程之间共享的事实.

我还添加了两个示例中增加线程池大小所需的线程池配置.默认值为10.

作为一种测试我刚刚说过的方法,你可以执行这个Python 3脚本(如果你现在没有python3,你有一个安装它的借口)它会在给定线程池的情况下或多或少地同时执行100个请求.

测试脚本

import time
import urllib.request
import concurrent.futures


URL = 'http://localhost:8080/'
TIMEOUT = 60
DELAY = 0.05
MAX_WORKERS = 20
REQ_RANGE = range(1,101)

def load_url():
    with urllib.request.urlopen(URL,timeout=TIMEOUT) as conn:
        return conn.read()


with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    futures = {}
    for i in REQ_RANGE:
        print("Sending req {}".format(i))
        futures[executor.submit(load_url)] = i
        time.sleep(DELAY)
    results = []
    for future in concurrent.futures.as_completed(futures):
        try:
            data = future.result().decode()
        except Exception as exc:
            print(exc)
        else:
            results.append((futures[future],data))
    curr_max = 0
    for i,data in sorted(results,key=lambda r: r[0]):
        new_max = int(data.split()[-1])
        assert new_max >= curr_max,"The data was not updated correctly"
        print("Req {}: {}".format(i,data))
        curr_max = new_max

根据日志确定您遇到问题的方式,这种问题不值得信任.特别是您无法控制请求在“访问”日志中记录的时间.我无法使用我的测试代码使代码失败但是在一般情况下确实存在竞争条件,在这个示例中它应该一直工作,因为代码只是生成atomic operation.只需要定期从一个属性分配中心点.

我希望代码是自我解释的,如果你有一个问题留下评论.

编辑：我编辑了基于插件的方法,因为它只能工作,因为只有一个地方正在执行插件,如果你创建另一个更新数据的后台任务,那么当你做的不仅仅是一个任务时,它可能会有问题.如果您将从一个BackgroundTask更新,无论代码是什么,您都可以找到它.