分享一个多线程/多线程任务定时调度项目 chronos

[lianghui]

调度由Tornado IOLoop负责，任务执行可为线程或者进程模式。

github 地址： https://github.com/thomashuang/chronos

安装 pip install chronospy

Hello World

import logging
import time
import tornado
import chronos
import os
import urllib2

def test_process():
    LOGGER.info("process pid %s", os.getpid())


def test(word):
    LOGGER.info("an other task, say '%s'", word)


def say():
    response = urllib2.urlopen('https://www.google.com/')
    html = response.read()
    LOGGER.info(html[:10])


def init():
    global LOGGER
    debug = True
    level = logging.DEBUG if debug else logging.INFO
    logging.basicConfig(level=level,
                        format='%(asctime)s %(levelname)-8s %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S', filemode='a+')

    LOGGER = logging.getLogger("demo")

    # bind a ioloop or use default ioloop
    chronos.setup()  # chronos.setup(tornado.ioloop.IOLoop())
    chronos.schedule('say', chronos.every_second(1), say)
    chronos.schedule('say2', chronos.every_second(1), test_process, once=True, process=True)
    chronos.schedule('say3', chronos.every_second(1), lambda: test("test3"))
    chronos.start(True)

if __name__ == '__main__':

    init()