在thread/threadpool/ioloop中运行函数的修饰符
threaded的Python项目详细描述
螺纹的
threaded是一组decorators，它将函数包装成：
- concurrent.futures.ThreadPool
- threading.Thread
- asyncio.Task in Python 3.
为什么？因为复制粘贴loop.create_task，threading.thread和thread_pool.submit很无聊， 尤其是当目标函数仅以这种方式使用时。
优点：
- 免费软件：apache许可证
- 开源：https://github.com/python-useful-helpers/threaded
- PYPI封装：https://pypi.python.org/pypi/threaded
- 测试：请参见顶部的百吉饼
- 支持多种python版本：
Python 3.4 Python 3.5 Python 3.6 Python 3.7 PyPy3 3.5+
注意
对于python 2.7/pypy，您可以使用1.x.x版
装饰工：
- 线程池-nativeconcurrent.futures.ThreadPool。
- threadpooled是threadpooled的别名。
- 线程化-包装成threading.Thread。
- threaded是threaded的别名。
- 异步任务-包装成asyncio.Task。使用与threadpooled相同的api。
- 异步任务是异步任务的别名。
用法
线程池
大多数情况下，它是必需的decorator:submit函数在调用时提交给threadpoolexecutor。
注意
python 3和python 2.7之间的api有很大不同。见下面的API部分。
threaded.ThreadPooled.configure(max_workers=3)
注意
默认情况下，如果未配置executor，则使用默认参数进行配置：max_workers=CPU_COUNT * 5
@threaded.ThreadPooleddeffunc():passconcurrent.futures.wait([func()])
python 3.5+与asyncio一起使用：
注意
如果循环getter不可调用，则忽略循环getter need上下文。
loop=asyncio.get_event_loop()@threaded.ThreadPooled(loop_getter=loop,loop_getter_need_context=False)deffunc():passloop.run_until_complete(asyncio.wait_for(func(),timeout))
python 3.5+使用从调用参数中提取异步和循环：
loop_getter=lambdatgt_loop:tgt_loop@threaded.ThreadPooled(loop_getter=loop_getter,loop_getter_need_context=True)# loop_getter_need_context is requireddeffunc(*args,**kwargs):passloop=asyncio.get_event_loop()loop.run_until_complete(asyncio.wait_for(func(loop),timeout))
在应用程序关闭期间，可以停止池（如果某个组件请求，池将自动重新创建）。
threaded.ThreadPooled.shutdown()
螺纹
经典的threading.Thread。用于运行直到关闭和自动关闭线程而不返回。
用法示例：
@threaded.Threadeddeffunc(*args,**kwargs):passthread=func()thread.start()thread.join()
如果没有参数，线程名将使用模式：'Threaded: ' + func.__name__
注意
如果函数名不可访问，则使用str（hash（func））。
重写名可以通过相应的参数don：
@threaded.Threaded(name='Function in thread')deffunc(*args,**kwargs):pass
线程可以自动守护：
@threaded.Threaded(daemon=True)deffunc(*args,**kwargs):pass
另外，如果在线程启动之前不需要任何添加操作， 可在返回前自动启动：
@threaded.Threaded(started=True)deffunc(*args,**kwargs):pass
异步任务
用asyncio.Task包装。
异步使用：
@threaded.AsyncIOTaskdeffunc():passloop=asyncio.get_event_loop()loop.run_until_complete(asyncio.wait_for(func(),timeout))
直接提供事件循环：
注意
如果循环getter不可调用，则忽略循环getter need上下文。
loop=asyncio.get_event_loop()@threaded.AsyncIOTask(loop_getter=loop)deffunc():passloop.run_until_complete(asyncio.wait_for(func(),timeout))
用于从调用参数中提取循环：
loop_getter=lambdatgt_loop:tgt_loop@threaded.AsyncIOTask(loop_getter=loop_getter,loop_getter_need_context=True)deffunc(*args,**kwargs):passloop=asyncio.get_event_loop()loop.run_until_complete(asyncio.wait_for(func(loop),timeout))
测试
软件包的主要测试机制是使用tox。 可用环境可通过tox-l
收集CI系统
对于代码检查，多个ci系统并行使用：
- Travis CI:用于检查：pep8、pylint、bandit、安装可能性和单元测试。同时它还出版了关于工作服的报道。
- coveralls:用于显示覆盖率。
- Azure CI:用于windows上的功能测试。
CD系统
Travis CI:用于pypi上的包传递。
欢迎加入QQ群-->： 979659372
