import threading
#自定义test函数
def test():
print("test threading")
#创建一个单线程,来执行test()函数
t = threading.Thread(target= test)
t.start()
|
import threading
from datetime import *
#自定义test函数
def test():
now = datetime.now() #获取当前时间
print("打印多线程执行时间:",now)
#自定义thr()函数,来执行多线程
def thr():
threads = [] #自定义一个空的数组,用来存放线程组
for i in range(10): #设置循环10次
t = threading.Thread(target=test)
threads.append(t) #把创建的的线程t,装到threads 数组中
#启动线程
for t in threads:
t.start()
#执行thr()函数进行多并发
if __name__ == "__main__" :
thr()
|
import threading
from datetime import *
#自定义test函数
def test():
now = datetime.now() #获取当前时间
print("打印多线程执行时间:",now)
#设置50个线程
def looptest():
for i in range(50):
test()
#自定义thr()函数,来执行多线程
def thr():
threads = [] #自定义一个空的数组,用来存放线程组
for i in range(20): #设置循环10次
t = threading.Thread(target=test)
threads.append(t) #把创建的的线程t,装到threads 数组中
#启动线程
for t in threads:
t.start()
#执行thr()函数进行多并发
if __name__ == "__main__" :
thr()
|
import threading
from datetime import *
#自定义test函数
def test():
x=0
while (x ==0): #修改成死循环
print(datetime.now())
#自定义thr()函数,来执行多线程
def thr():
threads = [] #自定义一个空的数组,用来存放线程组
for i in range(20): #设置循环10次
t = threading.Thread(target=test)
threads.append(t) #把创建的的线程t,装到threads 数组中
t.setDaemon(True) # 设置守护线程
#启动线程
for t in threads:
t.start()
#执行thr()函数进行多并发
if __name__ == "__main__" :
thr()
print("守护线程功能启用,end")
|
import threading
from datetime import *
def test():
x =0
while(x ==0):
print(datetime.now())
def thr():
threads = []
for i in range(10):
t = threading.Thread(target=test)
threads.append(t)
t.setDaemon(True)
for t in threads:
t.start()
for t in threads:
t.join()
if __name__ == "__main__":
thr()
print("阻塞线程功能启动,end")
|
for t in threads:
t.join(2)
|
# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ
@ time : 2020-6-8
"""
import threading
from datetime import *
#自定义全局变量需要的线程数,20
thread_num = 20
#自定义全局变量每个线程需要循环的数量,10
one_work_num = 10
#自定义函数test()
def test():
#编写测试代码
now = datetime.now()
print("打印时间戳:",now)
#设置死循环
#x =0
#while (x ==1):
# print("打印时间戳:",now)
def working():
# 引用全局变量
global one_work_num
#嵌套执行指定循环次数的 test()函数
for i in range(0,one_work_num):
test()
#自定义thr()函数,来执行多线程
def thr():
#自定义一个空的数组,用来存放线程组
threads = []
#设置循环次数为thread_num
for i in range(thread_num):
#将working()函数放入到线程中
t =threading.Thread(target=working,name="T" + str(i))
#设置守护线程
t.setDaemon(True)
threads.append(t)
#启动线程并执行
for t in threads:
t.start()
#设置阻塞线程
for t in threads:
t.join(2)
if __name__ == "__main__":
thr()
|
原文链接:http://www.51testing.com/html/20/n-4475920.html
热门关键词: 登录