Pythonでのバックグランド処理

GAEで重い処理をする際は、「deferred」というライブラリを使ってバックグラウンドで走らせていました。GAEを外すにあたり、何か代わりになるものを探しましたが見つからなかったので、pythonのthreadingを使って処理を書いてみました。ただ、deferredライブラリと違ってqueueを使っていません。対処しきれないようなアクセスが発生した場合、queueを使うように修正しようと考えています。また、まだ簡単にしかテストしていません。

バックグランド処理の例

バックグランド処理を受け持つクラスの作成

--- ファイル: lib/simple_thread.py ---

#!/usr/bin/env python
#coding: utf-8

import logging
import threading
import time
import traceback

def execute_logic(*args, **kwargs):
    t = SimpleThread(*args, **kwargs)
    t.start()


class SimpleThread(threading.Thread):
    # 繰り返し回数のデフォルト値を定義します。
    RETRY_TIMES = 10
    # 繰り返し間隔のデフォルト値を定義します。
    RETRY_INTERVAL = 10

    def __init__(self, *args, **kwargs):
        threading.Thread.__init__(self)
        # self._func = args[0]
        # self._args = args[1:]
        self._func, self._args = args[0], args[1:]
        self._retry_times = kwargs['retry_times'] if 'retry_times' 
                in kwargs else self.RETRY_TIMES
        self._retry_interval = kwargs['retry_interval'] 
                if 'retry_interval' in kwargs else self.RETRY_INTERVAL
        if logging.getLogger().isEnabledFor(logging.DEBUG):
            logging.debug('func: %s' % self._func)
            logging.debug('args: %s' % (self._args, ))
            logging.debug('retry_times: %s' % self._retry_times)
            logging.debug('retry_interval: %s' % self._retry_interval)

    def run(self):
        counter = 0
        while counter < self._retry_times:
            try:
                self._func(*self._args)
                break
            except TypeError, e:
                # TypeErrorは致命的なのでループを終了します。
                raise e
            except Exception, e:
                err_info = traceback.format_exc(e)
                logging.error(err_info)
                counter = counter + 1
                time.sleep(self._retry_interval)

バックグランドで実際に走らせるテストロジックの作成

--- ファイル: test_logic.py ---

#!/usr/bin/env python
#coding: utf-8

def test_sum(a, b):
    print a + b

テストするクラスの作成

--- ファイル: test.py ---

#!/usr/bin/env python
#coding: utf-8

import logging
import test_logic
from lib import simple_thread

def main():
    logging.getLogger().setLevel(logging.DEBUG)
    simple_thread.execute_logic(test_logic.test_sum, 10, 200,
                                retry_times=100, retry_interval=2)

if __name__ == '__main__':
    main()

実行結果

$ python test.py
DEBUG:root:func: <function test_sum at 0x1090d6938>
DEBUG:root:args: (10, 200)
DEBUG:root:retry_times: 100
DEBUG:root:retry_interval: 2
210

この記事が役に立った場合、シェアしていただけると励みになります!!