V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
爱意满满的作品展示区。
lianghui
V2EX  ›  分享创造

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

  •  
  •   lianghui ·
    whiteclover · 2015-07-27 22:04:50 +08:00 · 2586 次点击
    这是一个创建于 3412 天前的主题,其中的信息可能已经有所发展或是发生改变。

    调度由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()
    
    目前尚无回复
    关于   ·   帮助文档   ·   博客   ·   API   ·   FAQ   ·   实用小工具   ·   1218 人在线   最高记录 6679   ·     Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.5 · 21ms · UTC 18:31 · PVG 02:31 · LAX 10:31 · JFK 13:31
    Developed with CodeLauncher
    ♥ Do have faith in what you're doing.