简介:
Python 线程能够通过主线程,调用线程来执行其余命令,
为Python提供更不便的应用。
并发线程测试
# 命令调用形式import threading,time# 定义每个线程要运行的函数def run(n): print("task",n) time.sleep(1)# 生成一个线程实例target=指标,args=参数t1 = threading.Thread(target=run,args=("t1",))t2 = threading.Thread(target=run,args=("t2",))#启动线程t1.start()t2.start()
并发类的继承测试
import threading,time# 继承threading.Threadclass MyThread(threading.Thread): # 重写父类的构造函数 def __init__(self,n):# 重构构造函数 super(MyThread.self).__init__() self.n = n # 定义每个线程要运行的函数 def run(self): print("running task ",self.n)# 实例化 t1 = MyThread("t1")t2 = MyThread("t2")# 启动线程t1.start()t2.start()
实现并发串行执行命令
import threading,timedef run(n): print("task",n) time.sleep(1)start_time = time.time()for i in range(50): t = threading.Thread(target=run,args=("t-%s"%i,)) t.start() t.join() # 注:默认主线程不会等子线程执行完在执行。# 注:串行:期待前一个命令齐全执行完,在执行下一条命令。# 注:实例.join() #wait() 期待执行结束执行下条命令,可实现串行。
实现主线程期待子线程执行结束后在往下执行
import threading,time# 定义每个线程要运行的函数def run(n): print("task",n) time.sleep(2)# 测试执行后果后的用时工夫start_time = time.time()# 创立列表接管t.start()对象内容。t_objs = []# 循环50次for i in range(3): # 生成一个线程实例target=指标,args=参数 t = threading.Thread(target=run,args=("t-%s"%i,)) # 启动线程 调用run t.start() # 执行后果增加到列表 t_objs.append(t)# 将每个循环t对象一次进行期待。for t in t_objs: # 每循环一次期待一次 t.join()# 打印用时工夫print("cost:",time.time() - start_time)
- 守护过程
守护线程:依赖于主线程存在的,为主线程工作。
性能:设置成守护线程后,主线程敞开,守护线程也连带敞开。
应用场景:socketserver 每一个链接过去,socketserver就会启动一
个线程,当服务端敞开时,须要敞开其余线程,就能够用
到守护线程,敞开。
`
import threading,time
定义每个线程要运行的函数
def run(n):
print("task",n)time.sleep(5)
创立列表接管t.start()对象内容。
t_objs = []
循环3次
for i in range(3):
# 生成一个线程实例target=指标,args=参数t = threading.Thread(target=run,args=("t-%s"%i,))# 要在start之前设置# 把以后线程设置为守护线程t.setDaemon(True)# 启动线程 调用runt.start()
注:所有程序执行结束,还有一个join期待,执行结束退出程序。注:主线程MainThread,线程Thread。 threading.current_thread() #测试以后流动的线程类型。threading.active_count() #测试以后流动的线程数。