diff --git a/apps/common/decorators.py b/apps/common/decorators.py index 7e2c41ec5..38ce820b0 100644 --- a/apps/common/decorators.py +++ b/apps/common/decorators.py @@ -5,10 +5,8 @@ import functools import inspect import threading import time -import uuid from concurrent.futures import ThreadPoolExecutor -from django.core.cache import cache from django.db import transaction from .utils import logger @@ -46,27 +44,6 @@ def key_by_org(*args, **kwargs): return args[0].org_id -def _run_func_if_is_last(ttl, suffix_key, org, func, *args, **kwargs): - from orgs.utils import set_current_org - - try: - set_current_org(org) - uid = uuid.uuid4().__str__() - suffix_key_func = suffix_key if suffix_key else default_suffix_key - func_name = f'{func.__module__}_{func.__name__}' - key_suffix = suffix_key_func(*args, **kwargs) - key = f'DELAY_RUN_{func_name}_{key_suffix}' - cache.set(key, uid, ttl) - st = (ttl - 2 > 1) and ttl - 2 or 2 - time.sleep(st) - ret = cache.get(key, None) - - if uid == ret: - func(*args, **kwargs) - except Exception as e: - logger.error('delay run error: %s' % e) - - class LoopThread(threading.Thread): def __init__(self, loop, *args, **kwargs): super().__init__(*args, **kwargs) @@ -82,80 +59,61 @@ loop = asyncio.get_event_loop() loop_thread = LoopThread(loop) loop_thread.daemon = True loop_thread.start() -executor = ThreadPoolExecutor(max_workers=5, thread_name_prefix='debouncer') +executor = ThreadPoolExecutor(max_workers=10, + thread_name_prefix='debouncer') +_loop_debouncer_func_task_cache = {} +_loop_debouncer_func_args_cache = {} + + +def cancel_or_remove_debouncer_task(cache_key): + task = _loop_debouncer_func_task_cache.get(cache_key, None) + if not task: + return + if task.done(): + del _loop_debouncer_func_task_cache[cache_key] + else: + task.cancel() class Debouncer(object): - def __init__(self, callback, check, delay, *args, **kwargs): + def __init__(self, callback, check, delay, loop=None, executor=None): self.callback = callback self.check = check self.delay = delay + self.loop = loop + if not loop: + self.loop = asyncio.get_event_loop() + self.executor = executor async def __call__(self, *args, **kwargs): await asyncio.sleep(self.delay) - ok = await self._check(*args, **kwargs) + ok = await self._check() if ok: - await loop.run_in_executor(executor, self.callback, *args) + await self.loop.run_in_executor(self.executor, self.callback, *args) - async def _check(self, *args, **kwargs): + async def _check(self): if asyncio.iscoroutinefunction(self.check): - return await self.check(*args, **kwargs) - return await loop.run_in_executor(executor, self.check) + return await self.check() + return await self.loop.run_in_executor(self.executor, self.check) -def _run_func_with_org(org, func, *args, **kwargs): +def _run_func_with_org(key, org, func, *args, **kwargs): from orgs.utils import set_current_org - try: set_current_org(org) func(*args, **kwargs) except Exception as e: logger.error('delay run error: %s' % e) + _loop_debouncer_func_task_cache.pop(key, None) + _loop_debouncer_func_args_cache.pop(key, None) -def delay_run(ttl=5, key=None): +def delay_run(ttl=5, key=None, merge_args=False): """ 延迟执行函数, 在 ttl 秒内, 只执行最后一次 :param ttl: :param key: 是否合并参数, 一个 callback - :return: - """ - - def inner(func): - @functools.wraps(func) - def wrapper(*args, **kwargs): - from orgs.utils import get_current_org - org = get_current_org() - suffix_key_func = key if key else default_suffix_key - uid = uuid.uuid4().__str__() - func_name = f'{func.__module__}_{func.__name__}' - key_suffix = suffix_key_func(*args, **kwargs) - cache_key = f'DELAY_RUN_{func_name}_{key_suffix}' - # 延迟两倍时间，防止缓存过期，导致校验失败 - cache.set(cache_key, uid, ttl * 2) - - def _check_func(key_id, key_value): - ret = cache.get(key_id, None) - return key_value == ret - - check_func_partial = functools.partial(_check_func, cache_key, uid) - run_func_partial = functools.partial(_run_func_with_org, org, func) - asyncio.run_coroutine_threadsafe( - Debouncer(run_func_partial, check_func_partial, ttl)(*args, **kwargs), - loop=loop - ) - - return wrapper - - return inner - - -def merge_delay_run(ttl, key=None): - """ - 合并 func 参数，延迟执行, 在 ttl 秒内, 只执行最后一次 - func 参数必须是 *args - :param ttl: - :param key: 是否合并参数, 一个 callback + :param merge_args: 是否合并之前的参数, bool :return: """ @@ -164,39 +122,46 @@ def merge_delay_run(ttl, key=None): if len(sigs.parameters) != 1: raise ValueError('func must have one arguments: %s' % func.__name__) param = list(sigs.parameters.values())[0] - if not str(param).startswith('*'): - raise ValueError('func args must be startswith *: %s' % func.__name__) - + if not str(param).startswith('*') or param.kind == param.VAR_KEYWORD: + raise ValueError('func args must be startswith *: %s and not have **kwargs ' % func.__name__) suffix_key_func = key if key else default_suffix_key @functools.wraps(func) def wrapper(*args): - key_suffix = suffix_key_func(*args) + from orgs.utils import get_current_org + org = get_current_org() func_name = f'{func.__module__}_{func.__name__}' - cache_key = f'DELAY_MERGE_RUN_{func_name}_{key_suffix}' - values = cache.get(cache_key, []) - new_arg = [*values, *args] - cache.set(cache_key, new_arg, ttl) - return delay_run(ttl, suffix_key_func)(func)(*new_arg) + key_suffix = suffix_key_func(*args) + cache_key = f'DELAY_RUN_{func_name}_{key_suffix}' + new_arg = args + if merge_args: + values = _loop_debouncer_func_args_cache.get(cache_key, []) + new_arg = [*values, *args] + _loop_debouncer_func_args_cache[cache_key] = new_arg + + cancel_or_remove_debouncer_task(cache_key) + + run_func_partial = functools.partial(_run_func_with_org, cache_key, org, func) + _debouncer = Debouncer(run_func_partial, lambda: True, ttl, + loop=loop, executor=executor) + task = asyncio.run_coroutine_threadsafe(_debouncer(*new_arg), + loop=loop) + _loop_debouncer_func_task_cache[cache_key] = task return wrapper return inner +merge_delay_run = functools.partial(delay_run, merge_args=True) + + @delay_run(ttl=5) -def test_delay_run(username, year=2000): - print("Hello, %s, now is %s" % (username, year)) +def test_delay_run(*username): + print("Hello, %s, now is %s" % (username, time.time())) -@merge_delay_run(ttl=5, key=lambda *users: users[0][0]) -def test_merge_delay_run(*users): - name = ','.join(users) - time.sleep(2) - print("Hello, %s, now is %s" % (name, time.time())) - - -@merge_delay_run(ttl=5, key=lambda *users: users[0][0]) +@delay_run(ttl=5, key=lambda *users: users[0][0], merge_args=True) def test_merge_delay_run(*users): name = ','.join(users) time.sleep(2) @@ -210,6 +175,7 @@ def do_test(): # test_delay_run('test', year=i) test_merge_delay_run('test %s' % i) test_merge_delay_run('best %s' % i) + test_delay_run('test run %s' % i) end = time.time() using = end - s