1 Star 0 Fork 10

fangjiaqi77/薅羊毛

forked from hotootop/薅羊毛 
加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
该仓库未声明开源许可证文件(LICENSE),使用请关注具体项目描述及其代码上游依赖。
克隆/下载
mythread.py 1.66 KB
一键复制 编辑 原始数据 按行查看 历史
hotootop 提交于 2021-02-02 11:09 . 主要代码
import threading
import time
from time import sleep
import sys
import atexit
class MyTimer(threading.Thread):
def __init__(self, seconds, mytarget, myargs=None, *args, **kwargs):
super().__init__(*args, **kwargs)
self.__flag = threading.Event() # 用于暂停线程的标识
self.__flag.set() # 设置为True
self.__running = threading.Event() # 用C于停止线程的标识
self.__running.set() # 将running设置为True
self.seconds = seconds
self.mytarget = mytarget
self.myargs = myargs
def run(self):
while self.__running.isSet():
self.__flag.wait() # 为True时立即返回, 为False时阻塞直到内部的标识位为True后返回
sleep(self.seconds)
if self.myargs:
self.mytarget(*self.myargs)
else:
self.mytarget()
def pause(self):
self.__flag.clear() # 设置为False, 让线程阻塞
def resume(self):
self.__flag.set() # 设置为True, 让线程停止阻塞
def stop(self):
print('stop')
self.__flag.set() # 将线程从暂停状态恢复, 如何已经暂停的话
self.__running.clear() # 设置为False
if __name__ == '__main__':
def test():
print('test')
mythread = MyTimer(2, test, daemon=True)
mythread.start()
atexit.register(mythread.stop)
sleep(3)
mythread.pause()
print('暂停')
sleep(3)
mythread.resume()
print('恢复')
sleep(3)
print('结束')
sys.exit()
print('test')
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
Python
1
https://gitee.com/fangjiaqi77/collect-wool.git
git@gitee.com:fangjiaqi77/collect-wool.git
fangjiaqi77
collect-wool
薅羊毛
master

搜索帮助