神器celery 源码解析 - 4

# 根据定义递归求解
def fib(n):
    if n <= 1: 
        return n
    return fib(n - 1) + fib(n - 2)
复制代码

f(5)=f(4)                    +f(3)
    =f(3)          +f(2)     +f(2)     +f(1)
    =f(2)     +f(1)+f(1)+f(0)+f(1)+f(0)+f(1)
    =f(1)+f(0)+f(1)+f(1)+f(0)+f(1)+f(0)+f(1)
    =5
复制代码

# 根据定义递归求解
cache = {}
def fib_v1(n):
    if n in cache:
        return cache[n]
    if n <= 1: 
        result = n
    else:
        result =  fib(n - 1) + fib(n - 2)
    cache[n] = result
    return result
复制代码

def cache_decorator(fun):
    _cache = {}
    def wrapper(*args, **kwargs):
        if args in _cache:
            return _cache[args]
        else:
            ret = fun(*args, **kwargs)
            _cache[args] = ret
            return ret
    return wrapper
复制代码

@cache_decorator
def fib(n):
    ...
复制代码

class LRUCache(abc.MutableMapping, Generic[K, V]):
    def __init__(self, capacity=None):
        self.capacity = capacity  # 缓存容量
        self.cache = OrderedDict()  # 有序字典
    def get(self, key: K, default: D = None) -> Optional[Union[V, D]]:
        value = self.cache.get(key)  # 从换成获取
        if value is not None:
            del self.cache[key]
            self.cache[key] = value  # 更新缓存顺序
            return value
        return default
    def set(self, key: K, value: V):
        if self.cache.get(key):
            del self.cache[key]  
            self.cache[key] = value # 更新缓存顺序及值
        else:
            self.cache[key] = value
            if self.capacity is not None and self.length > self.capacity:
                self.cache.popitem(last=False)  # 淘汰最古老的数据
复制代码

from functools import lru_cache
@lru_cache()
def fib(n):
    ...
复制代码

# kombu-5.0.0/kombu/utils/functional.py
class LRUCache(UserDict):
    """LRU Cache implementation using a doubly linked list to track access.
    """
    def __init__(self, limit=None):
        self.limit = limit
        self.mutex = threading.RLock()
        self.data = OrderedDict()
    def __getitem__(self, key):
        with self.mutex:
            value = self[key] = self.data.pop(key)
            return value
    def __setitem__(self, key, value):
        # remove least recently used key.
        with self.mutex:
            if self.limit and len(self.data) >= self.limit:
                self.data.pop(next(iter(self.data)))
            self.data[key] = value
    ...
复制代码

def memoize(maxsize=None, keyfun=None, Cache=LRUCache):
    """Decorator to cache function return value."""
    def _memoize(fun):
        mutex = threading.Lock()
        cache = Cache(limit=maxsize)
        @wraps(fun)
        def _M(*args, **kwargs):
            if keyfun:
                key = keyfun(args, kwargs)
            else:
                key = args + (KEYWORD_MARK,) + tuple(sorted(kwargs.items()))
            try:
                with mutex:
                    value = cache[key]
            except KeyError:
                value = fun(*args, **kwargs)
                # 未命中需要执行函数
                _M.misses += 1
                with mutex:
                    cache[key] = value
            else:
                # 命中率增加
                _M.hits += 1
            return value
        def clear():
            """Clear the cache and reset cache statistics."""
            # 清理缓存及统计
            cache.clear()
            _M.hits = _M.misses = 0
        # 统计信息
        _M.hits = _M.misses = 0
        _M.clear = clear
        _M.original_func = fun
        return _M
    return _memoize
复制代码

class TokenBucket:
    #: The rate in tokens/second that the bucket will be refilled.
    fill_rate = None
    #: Maximum number of tokens in the bucket.
    capacity = 1
    #: Timestamp of the last time a token was taken out of the bucket.
    timestamp = None
    def __init__(self, fill_rate, capacity=1):
        # 容量上限
        self.capacity = float(capacity)
        # 剩余令牌数，初始等于容量上限
        self._tokens = capacity
        # 填充率
        self.fill_rate = float(fill_rate)
        self.timestamp = monotonic()
        # 数据容器
        self.contents = deque()
    def add(self, item):
        self.contents.append(item)
    def pop(self):
        # 先进先出
        return self.contents.popleft()
复制代码

def can_consume(self, tokens=1):
    """Check if one or more tokens can be consumed.
    Returns:
        bool: true if the number of tokens can be consumed
            from the bucket.  If they can be consumed, a call will also
            consume the requested number of tokens from the bucket.
            Calls will only consume `tokens` (the number requested)
            or zero tokens -- it will never consume a partial number
            of tokens.
    """
    if tokens <= self._get_tokens():
        # 消费n个令牌
        self._tokens -= tokens
        return True
    return False
def _get_tokens(self):
    if self._tokens < self.capacity:
        # 记录当前时间
        now = monotonic()
        # 计算已经流失的令牌数量
        delta = self.fill_rate * (now - self.timestamp)
        # 更新容量上限或者剩余令牌和流失数量之和
        self._tokens = min(self.capacity, self._tokens + delta)
        self.timestamp = now
    return self._tokens
复制代码

#ch23-celery/kombu-5.0.0/kombu/mixins.py:240
class ConsumerMixin:
    def run(self, _tokens=1, **kwargs):
        restart_limit = TokenBucket(1)
        ...
        # 无限循环
        while not self.should_stop:
            try:
                # 有令牌消费
                if restart_limit.can_consume(_tokens):  # pragma: no cover
                    for _ in self.consume(limit=None, **kwargs):
                        pass
                else:
                    # 没浪费休眠
                    sleep(restart_limit.expected_time(_tokens))
            except errors:
                ...
复制代码

def expected_time(self, tokens=1):
    """Return estimated time of token availability.
    Returns:
        float: the time in seconds.
    """
    _tokens = self._get_tokens()
    tokens = max(tokens, _tokens)
    return (tokens - _tokens) / self.fill_rate
复制代码

class round_robin_cycle:
    """Iterator that cycles between items in round-robin."""
    """轮询调度算法"""
    def __init__(self, it=None):
        self.items = it if it is not None else []
    def update(self, it):
        """Update items from iterable."""
        """更新列表"""
        self.items[:] = it
    def consume(self, n):
        """Consume n items."""
        """消费n个元素"""
        return self.items[:n]
    def rotate(self, last_used):
        """Move most recently used item to end of list."""
        """旋转:把最后一个元素放到列表某尾"""
        items = self.items
        try:
            items.append(items.pop(items.index(last_used)))
        except ValueError:
            pass
        return last_used
复制代码

def test_round_robin_cycle():
    it = cycle_by_name('round_robin')(['A', 'B', 'C'])
    assert it.consume(3) == ['A', 'B', 'C']
    it.rotate('B')
    assert it.consume(3) == ['A', 'C', 'B']
    it.rotate('A')
    assert it.consume(3) == ['C', 'B', 'A']
    it.rotate('A')
    assert it.consume(3) == ['C', 'B', 'A']
    it.rotate('C')
    assert it.consume(3) == ['B', 'A', 'C']
复制代码

class FairCycle:
    """Cycle between resources.
    Consume from a set of resources, where each resource gets
    an equal chance to be consumed from.
    Arguments:
        fun (Callable): Callback to call.
        resources (Sequence[Any]): List of resources.
        predicate (type): Exception predicate.
    """
    def __init__(self, fun, resources, predicate=Exception):
        self.fun = fun
        self.resources = resources
        self.predicate = predicate
        #  初始位置
        self.pos = 0
复制代码

def _next(self):
    while 1:
        try:
            resource = self.resources[self.pos]
            # 位置加1
            self.pos += 1
            return resource
        except IndexError:
            # 到尾部后，重置位置
            self.pos = 0
            if not self.resources:
                raise self.predicate()
def get(self, callback, **kwargs):
    """Get from next resource."""
    # 无限重试
    for tried in count(0):  # for infinity
        # 获取资源
        resource = self._next()
        try:
            # 利用资源
            return self.fun(resource, callback, **kwargs)
        except self.predicate:
            # reraise when retries exchausted.
            # 容错上限
            if tried >= len(self.resources) - 1:
                raise
复制代码

class LamportClock:
    #: The clocks current value.
    value = 0
    def __init__(self, initial_value=0, Lock=Lock):
        self.value = initial_value
        self.mutex = Lock()
    def adjust(self, other):
        with self.mutex:
            value = self.value = max(self.value, other) + 1
            return value
    def forward(self):
        with self.mutex:
            self.value += 1
            return self.value
复制代码

>>> from future import print_statement as printf
# ^ ignore: just fooling stupid pyflakes
>>> x = LaxBoundedSemaphore(2)
>>> x.acquire(printf, 'HELLO 1')
HELLO 1
>>> x.acquire(printf, 'HELLO 2')
HELLO 2
>>> x.acquire(printf, 'HELLO 3')
>>> x._waiters   # private, do not access directly
[print, ('HELLO 3',)]
>>> x.release()
HELLO 3
复制代码

class LaxBoundedSemaphore:
    def __init__(self, value):
        # 信号容量
        self.initial_value = self.value = value
        # 使用双端队列，FIFO
        self._waiting = deque()
        self._add_waiter = self._waiting.append
        self._pop_waiter = self._waiting.popleft
复制代码

def acquire(self, callback, *partial_args, **partial_kwargs):
    """Acquire semaphore.
    This will immediately apply ``callback`` if
    the resource is available, otherwise the callback is suspended
    until the semaphore is released.
    Arguments:
        callback (Callable): The callback to apply.
        *partial_args (Any): partial arguments to callback.
    """
    value = self.value
    if value <= 0:
        # 容量不够的时候先暂存执行函数，并不更改可用数量
        self._add_waiter((callback, partial_args, partial_kwargs))
        return False
    else:
        # 可用数量-1
        self.value = max(value - 1, 0)
        # 直接执行函数
        callback(*partial_args, **partial_kwargs)
        return True
复制代码

def release(self):
    """Release semaphore.
    Note:
        If there are any waiters this will apply the first waiter
        that is waiting for the resource (FIFO order).
    """
    try:
        waiter, args, kwargs = self._pop_waiter()
    except IndexError:
        # 无缓存则只增加可用数量
        self.value = min(self.value + 1, self.initial_value)
    else:
        # 有缓存则执行第一个缓存，可用数量不变还是小于0
        waiter(*args, **kwargs)
复制代码

# kombu-5.0.0/kombu/utils/functional.py
def retry_over_time(fun, catch, args=None, kwargs=None, errback=None,
                    max_retries=None, interval_start=2, interval_step=2,
                    interval_max=30, callback=None, timeout=None):
    kwargs = {} if not kwargs else kwargs
    args = [] if not args else args
    interval_range = fxrange(interval_start,
                             interval_max + interval_start,
                             interval_step, repeatlast=True)
    # 超时时间
    end = time() + timeout if timeout else None
    for retries in count():
        try:
            return fun(*args, **kwargs)
        except catch as exc:
            # 超过次数
            if max_retries is not None and retries >= max_retries:
                raise
            # 超过时间
            if end and time() > end:
                raise
            ...
            # 休眠
            sleep(1.0)
复制代码