chronospy定時任務(wù)模塊

Chronos 是通過 Tornado IOLoop 驅(qū)動的多線程/多進程任務(wù)定時工具。

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()