摘要：最近剛看完多線程，為了加深印象，按照分鐘實(shí)現(xiàn)延遲消息功能的思路，實(shí)現(xiàn)了一個(gè)簡(jiǎn)易版的異步隊(duì)列。讀取任務(wù)時(shí)，計(jì)算當(dāng)前和，取出需要執(zhí)行的任務(wù)，使用多線程的形式執(zhí)行。加鎖的主要作用是防止多線程同時(shí)操作文件讀寫(xiě)，影響數(shù)據(jù)一致性。

最近剛看完python多線程，為了加深印象，按照1分鐘實(shí)現(xiàn)“延遲消息”功能的思路，實(shí)現(xiàn)了一個(gè)簡(jiǎn)易版的異步隊(duì)列。

高效延時(shí)消息，包含兩個(gè)重要的數(shù)據(jù)結(jié)構(gòu)：
1.環(huán)形隊(duì)列，例如可以創(chuàng)建一個(gè)包含3600個(gè)slot的環(huán)形隊(duì)列（本質(zhì)是個(gè)數(shù)組）
2.任務(wù)集合，環(huán)上每一個(gè)slot是一個(gè)Set

同時(shí)，啟動(dòng)一個(gè)timer，這個(gè)timer每隔1s，在上述環(huán)形隊(duì)列中移動(dòng)一格，有一個(gè)Current Index指針來(lái)標(biāo)識(shí)正在檢測(cè)的slot。
Task結(jié)構(gòu)中有兩個(gè)很重要的屬性：
（1）Cycle-Num：當(dāng)Current Index第幾圈掃描到這個(gè)Slot時(shí)，執(zhí)行任務(wù)
（2）Task-Function：需要執(zhí)行的任務(wù)指針

下邊是代碼（代碼不止100行，但是在200行內(nèi)，也算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 的數(shù)據(jù)結(jié)構(gòu)
# 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

# 為了防止任務(wù)太多需要生成過(guò)多的線程，我們使用Queue 來(lái)限制生成的線程數(shù)量
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):
    # 使用鎖來(lái)保證每次只要一個(gè)線程寫(xiě)入文件，防止數(shù)據(jù)出錯(cuò)
    with lock:
        with open(tasks_file, "r+") as json_file:
            count_down = kwargs.pop("count_down", 0)
            tasks = json.load(json_file)
            # 執(zhí)行時(shí)間
            exec_time = int(time()) + count_down
            # 循環(huán)索引
            index = str(exec_time % 3600)
            # 圈數(shù)
            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)
        # 執(zhí)行時(shí)間
        current_time = int(time())
        # 循環(huán)索引
        index = str(current_time % 3600)
        # 圈數(shù)
        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分鐘實(shí)現(xiàn)“延遲消息”功能的思路。隊(duì)列的數(shù)據(jù)結(jié)構(gòu)為

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

index的值為 1-3600。每小時(shí)一個(gè)循環(huán)。
cycle_num 則是 由 (時(shí)間戳 / 3600 + 1) 計(jì)算得到的值，是圈數(shù)。

每當(dāng)有任務(wù)加入，我們計(jì)算出index和cycle_num 將參數(shù)和方法名寫(xiě)入json文件。
讀取任務(wù)時(shí)，計(jì)算當(dāng)前 index和cycle_num， 取出需要執(zhí)行的任務(wù)，使用多線程的形式執(zhí)行。

為了防止任務(wù)太多需要生成過(guò)多的線程，我們使用Queue 來(lái)限制生成的線程數(shù)量。

加鎖的主要作用是防止多線程同時(shí)操作文件讀寫(xiě)，影響數(shù)據(jù)一致性。

當(dāng)然，也可以使用redis 存儲(chǔ)隊(duì)列，因?yàn)?redis 是單線程操作，可以防止多線程操作影響數(shù)據(jù)一致性的問(wèn)題。
這一部分有需要的可以自己實(shí)現(xiàn)。

參考：

python線程筆記

1分鐘實(shí)現(xiàn)“延遲消息”功能