摘要：最近剛看完多線程，為了加深印象，按照分鐘實現延遲消息功能的思路，實現了一個簡易版的異步隊列。讀取任務時，計算當前和，取出需要執行的任務，使用多線程的形式執行。加鎖的主要作用是防止多線程同時操作文件讀寫，影響數據一致性。

最近剛看完python多線程，為了加深印象，按照1分鐘實現“延遲消息”功能的思路，實現了一個簡易版的異步隊列。

高效延時消息，包含兩個重要的數據結構：
1.環形隊列，例如可以創建一個包含3600個slot的環形隊列（本質是個數組）
2.任務集合，環上每一個slot是一個Set

同時，啟動一個timer，這個timer每隔1s，在上述環形隊列中移動一格，有一個Current Index指針來標識正在檢測的slot。
Task結構中有兩個很重要的屬性：
（1）Cycle-Num：當Current Index第幾圈掃描到這個Slot時，執行任務
（2）Task-Function：需要執行的任務指針

下邊是代碼（代碼不止100行，但是在200行內，也算100行了。）

#! -*- coding: utf-8 -*-

try:
    import cPickle as pickle
except ImportError:
    import pickle
try:
    import simplejson as json
except ImportError:
    import json

import os
import errno
import Queue
import random
import logging
from functools import wraps
from threading import Timer, RLock, Thread
from time import sleep, time
from base64 import b64encode, b64decode

# json 的數據結構
# tasks = {
#     index: {
#         cycle_num: [(func, bargs)]
#     }
# }

logging.basicConfig(level=logging.DEBUG,
                    format="(%(asctime)-15s) %(message)s",)
tasks_file = "tasks.json"
flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY

# 為了防止任務太多需要生成過多的線程，我們使用Queue 來限制生成的線程數量
WORKER_NUMS = 2
q = Queue.Queue(WORKER_NUMS)

lock = RLock()


def check_file():
    try:
        file_handle = os.open(tasks_file, flags)
    except OSError as e:
        if e.errno == errno.EEXIST:  # Failed as the file already exists.
            pass
        else:
            raise
    else:
        with os.fdopen(file_handle, "w") as file_obj:
            file_obj.write("{}")


def set_delay_task(func_name, *args, **kwargs):
    # 使用鎖來保證每次只要一個線程寫入文件，防止數據出錯
    with lock:
        with open(tasks_file, "r+") as json_file:
            count_down = kwargs.pop("count_down", 0)
            tasks = json.load(json_file)
            # 執行時間
            exec_time = int(time()) + count_down
            # 循環索引
            index = str(exec_time % 3600)
            # 圈數
            cycle_num = str(exec_time / 3600 + 1)
            dargs = pickle.dumps((args, kwargs))
            bargs = b64encode(dargs)
            index_data = tasks.get(index, {})
            index_data.setdefault(cycle_num, []).append((func_name, bargs))
            tasks[index] = index_data
            json_file.seek(0)
            json.dump(tasks, json_file)
            logging.debug("Received task: %s" % func_name)


def get_delay_tasks():
    with open(tasks_file, "r+") as json_file:
        tasks = json.load(json_file)
        # 執行時間
        current_time = int(time())
        # 循環索引
        index = str(current_time % 3600)
        # 圈數
        cycle_num = str(current_time / 3600 + 1)
        current_tasks = tasks.get(index, {}).get(cycle_num, [])
    tasks = []
    for func, bargs in current_tasks:
        dargs = b64decode(bargs)
        args, kwargs = pickle.loads(dargs)
        tasks.append((func, (args, kwargs)))
    return tasks


def get_method_by_name(method_name):
    possibles = globals().copy()
    possibles.update(locals())
    method = possibles.get(method_name)
    return method


def create_task(task_class, func, task_name=None, **kwargs):

    def execute(self):
        args, kwargs = self.data or ((), {})
        return func(*args, **kwargs)

    attrs = {
        "execute": execute,
        "func_name": func.__name__,
        "__module__": func.__module__,
        "__doc__": func.__doc__
    }
    attrs.update(kwargs)

    klass = type(
        task_name or func.__name__,
        (task_class,),
        attrs
    )

    return klass


class Hu(object):

    def __init__(self, func_name=None):
        self.func_name = func_name
        check_file()

    def task(self):
        def deco(func):
            self.func_name = func.__name__
            klass = create_task(Hu, func, self.func_name)
            func.delay = klass(func_name=klass.func_name).delay
            @wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)
            return wrapper
        return deco

    def delay(self, *args, **kwargs):
        _args = [self.func_name]
        _args.extend(args)
        Timer(0, set_delay_task, _args, kwargs).start()
        return True


def boss():
    while True:
        current_tasks = get_delay_tasks()
        for func, params in current_tasks:
            # Task accepted: auth.tasks.send_msg
            logging.debug("Task accepted: %s" % func)
            q.put((func, params))
        sleep(1)


def worker():
    while True:
        func, params = q.get()
        print "get task: %s
" % func
        method = get_method_by_name(func)
        args, kwargs = params
        # Task auth.tasks.send_msgsucceeded in
        start_time = time()
        method(*args, **kwargs)
        end_time = time()
        logging.debug("Task %s succeeded in %s" % (str(func), end_time - start_time))
        q.task_done()


def main():
    check_file()
    print("starting at:", time())
    for target in (boss, worker):
        t = Thread(target=target)
        t.start()
    print("all DONE at:", time())

hu = Hu()

# 使用方式如下：

@hu.task()
def test(num):
    sleep(2)
    print "test: %s" % num


if __name__ == "__main__":
    for i in range(10):
        test.delay(i, count_down=random.randint(1, 10))
    main()

# output

(2017-03-21 15:59:20,394) Received task: test
(2017-03-21 15:59:20,396) Received task: test
(2017-03-21 15:59:20,397) Received task: test
(2017-03-21 15:59:20,398) Received task: test
(2017-03-21 15:59:20,400) Received task: test
(2017-03-21 15:59:20,401) Received task: test
(2017-03-21 15:59:20,403) Received task: test
(2017-03-21 15:59:20,404) Received task: test
(2017-03-21 15:59:20,406) Received task: test
(2017-03-21 15:59:20,408) Received task: test
get task: test

(2017-03-21 15:59:21,395) Task accepted: test
(2017-03-21 15:59:22,397) Task accepted: test
(2017-03-21 15:59:22,397) Task accepted: test
(2017-03-21 15:59:22,397) Task accepted: test
test: 2
get task: test

(2017-03-21 15:59:23,399) Task test succeeded in 2.0037419796
(2017-03-21 15:59:24,404) Task accepted: test
test: 1
get task: test

按照1分鐘實現“延遲消息”功能的思路。隊列的數據結構為

{
    index: {
        cycle_num: [(func, bargs)]
    }
}

index的值為 1-3600。每小時一個循環。
cycle_num 則是 由 (時間戳 / 3600 + 1) 計算得到的值，是圈數。

每當有任務加入，我們計算出index和cycle_num 將參數和方法名寫入json文件。
讀取任務時，計算當前 index和cycle_num， 取出需要執行的任務，使用多線程的形式執行。

為了防止任務太多需要生成過多的線程，我們使用Queue 來限制生成的線程數量。

加鎖的主要作用是防止多線程同時操作文件讀寫，影響數據一致性。

當然，也可以使用redis 存儲隊列，因為 redis 是單線程操作，可以防止多線程操作影響數據一致性的問題。
這一部分有需要的可以自己實現。

參考：

python線程筆記

1分鐘實現“延遲消息”功能