线程与进程入门整理
多任务概念引入
#正常的程序执行
from time import sleep import time def sing(): for i in range(3): print("正在唱歌%d"%i) sleep(1) def dance(): for i in range(3): print('正在跳舞%d'%i) sleep(1) if __name__=='__main__': start_time=time.time() sing() dance()
正在唱歌0 正在唱歌1 正在唱歌2 正在跳舞0 正在跳舞1 正在跳舞2 6.007343530654907
引入并发的概念（多任务）
#在唱歌时调用跳舞
#多任务的概念引入(不考虑时间)
#正常执行程序
from time import sleep
import time
def sing():
for i in range(3):
print("正在唱歌%d"%i)
dance(i)
sleep(1)
def dance(i):
print('正在跳舞%d'%i)
sleep(1)
if __name__=='__main__':
start_time=time.time()
sing()
正在唱歌0 正在跳舞0 正在唱歌1 正在跳舞1 正在唱歌2 正在跳舞2
进程使用
#multiprocessing #subprocess多进程包
a、引入multiprocess中的Process
#创建子进程并执行 （注意：jupyter只能跟踪主进程，没法跟踪子进程。）
from multiprocessing import Process
def run_test():
print('执行子进程')
if __name__=='__main__':
print('主进程执行')
#创建子进程target接收执行的任务
p=Process(target=run_test)
#调用子进程
p.start()#这里实际上已经执行
"""
主进程执行
执行子进程
"""
b、给子进程函数传递参数
和使用 thread 类创建子线程的方式非常类似，使用 Process 类创建实例化对象，其本质是调用该类的构造方法创建新进程。Process 类的构造方法格式如下：
1)Process
Process(target=func_name,args=(,,),kwargs={key:value}) 调用 start() 方法启动该进程
form multiprocess import Process
from time import sleep
def run_test(name,**kwargs):
print('执行子进程那么值为%s'%name)
print("字典传入的值为",kwargs)
if __name__=="__main__":
print('主进程开始执行')
#创建子进程
p=Process(target=run_test,args=('test',),kwargs={'key':12}) #这里逗号不可省去
#调用子进程
p.start()
"""
主进程开始执行
执行子进程那么值为test
字典传入的值为 {'key': 12}
"""
需要说明的是，通过 multiprocessing.Process 来创建并启动进程时，程序必须先判断 if __name__=='__main__':，否则运行该程序会引发异常。
4)通过os.getpid比对主进程与新建进程的关系
from multiprocessing import Process
import os
def get_process(name,*args):
print(name)
for i in args:
print('当前执行{0},进程id为{1}'.format(i,os.getpid()))
if __name__=='__main__':
#创建一个进程
tuple_obj=(
'test_url1',
'test_url2',
'test_url2'
)
p=Process(target=get_process,args=('myCreadtProcess',*tuple_obj))
p.start()#开始运行进程
#p.join()
get_process('主进程',*tuple_obj)
主进程 当前执行test_url1,进程id为16676 当前执行test_url2,进程id为16676 当前执行test_url2,进程id为16676 myCreadtProcess 当前执行test_url1,进程id为3336 当前执行test_url2,进程id为3336 当前执行test_url2,进程id为3336
c.Process进程常用属性和方法
| 属性名或方法名 | 功能 |
|---|---|
| run() | 第 2 种创建进程的方式需要用到，继承类中需要对方法进行重写，该方法中包含的是新进程要执行的代码。 |
| start() | 和启动子线程一样，新创建的进程也需要手动启动，该方法的功能就是启动新创建的线程。 |
| join([timeout]) | 和 thread 类 join() 方法的用法类似，其功能是在多进程执行过程，其他进程必须等到调用 join() 方法的进程执行完毕（或者执行规定的 timeout 时间）后，才能继续执行； |
| is_alive() | 判断当前进程是否还活着。 |
| terminate() | 中断该进程。 |
| name属性 | 可以为该进程重命名，也可以获得该进程的名称。 |
| daemon | 和守护线程类似，通过设置该属性为 True，可将新建进程设置为“守护进程”。 |
| pid | 返回进程的 ID 号。大多数操作系统都会为每个进程配备唯一的 ID 号。 |
1）、join([timeout]) 等待进程p终止。Timeout是可选的超时期限，
#Process进程的实例方法
#join([timeout]) 等待进程p终止。Timeout是可选的超时期限，
#join 的使用
#未调用join时：
#由于子进程中有sleep()主进程并不会等子进程执行完成再执行，因此引入join概念
from multiprocessing import Process
from time import sleep
def worker(interval):
print('子进程开始执行')
sleep(interval)
print('子进程执行完')
if __name__=='__main__':
print('主进程正在执行')
#创建子进程
p=Process(target=worker,args=(3,))
#调用子进程
p.start()
print("主进程执行完")
"""
主进程正在执行
主进程执行完
子进程开始执行
子进程执行完
#由于子进程中有sleep()主进程并不会等子进程执行完成再执行，因此引入join概念
"""
#调用join方法时：
#主进程会调用子进程
#加入进程join属性主进程会等子进程执行完再执行
def worker(interval):
print('子进程开始执行')
sleep(interval)
print('子进程执行完')
if __name__=='__main__':
print('主进程正在执行')
#创建子进程
p=Process(target=worker,args=(3,))
#调用子进程
p.start()
p.join()
print("主进程执行完")
"""
主进程正在执行
子进程开始执行
子进程执行完
主进程执行完
"""
#jion(）方法中timeout的参数
#timeout时间与sleep时间之间比对影响进程执行
#由于设定让主进程等待时间为2秒，而子进程时间超过改时间，故主进程先行执行完
def worker(interval):
print('子进程开始执行')
sleep(interval)
print('子进程执行完')
if __name__=='__main__':
print('主进程正在执行')
#创建子进程
p=Process(target=worker,args=(5,))
#调用子进程
p.start()
p.join(time=2)
print("主进程执行完")
"""
主进程正在执行
子进程开始执行
主进程执行完
子进程执行完
"""
#没有调用join方法时，p的参数运行情况
import multiprocessing
import time
def clock(interval):
for i in range(3):
print('当前时间{}'.format(time.ctime()))
time.sleep(interval)
if __name__=='__main__':
#创建子进程
p=multiprocessing.Process(target=clock,args=(1,))
p.start()
#p.join()这里注释掉join方法，name主进程直接运行，不等待子进程结束完成
print('p.pid:',p.pid)
print('p.name:',p.name)
print('p.is_alive:',p.is_alive())#进程是否结束
"""
p.pid: 5596
p.name: Process-1
p.is_alive: True #子进程未结束
当前时间：Sat Oct 31 11:47:38 2020
当前时间：Sat Oct 31 11:47:39 2020
当前时间：Sat Oct 31 11:47:40 2020
"""
2)、属性 #p,name,p.pid
#Process 实例属性表
#p,name,p.pid
"""
方法 描述
name 进程的名称
pid 进程的整数进程
"""
#p的两个属性
import multiprocessing
import time
def clock(interval):
for i in range(3):
print('当前时间{}'.format(time.ctime()))
time.sleep(interval)
if __name__=='__main__':
#创建子进程
p=multiprocessing.Process(target=clock,args=(1,))
p.start()
p.join()
print('p.pid:',p.pid)
print('p.name:',p.name)
print('p.is_alive:',p.is_alive())#进程是否结束
"""
这里的三个print是子进程，由于有join方法的存在，
因此print会在子进程结束后在运行
因此这里的is_alive()返回false
当前时间：Sat Oct 31 11:32:59 2020
当前时间：Sat Oct 31 11:33:00 2020
当前时间：Sat Oct 31 11:33:01 2020
p.pid: 15532
p.name: Process-1
p.is_alive: False
"""
D、测试多个进程
p1=Process()
p2=Process()
p3=Process()
#测试多个进程
#创建函数并将其作为多个进程
def work1(interval):
print('work1')
sleep(interval)
print('end work1')
def work2(interval):
print('work2')
sleep(interval)
print('end work2')
def work3(interval):
print('work3')
sleep(interval)
print('end work3')
if __name__=='__main__':
print('执行主程序')
p1=Process(target=work1,args=(4,)) #这里4是int输入字符串出错
p2=Process(target=work2,args=(3,))
p3=Process(target=work3,args=(2,))
p1.start()
p2.start()
p3.start()
p1.join()
p2.join()
p3.join()
print('p1.name:', p1.name)
print('p2.name:', p2.name)
print('p3.name:', p3.name)
print('主进程执行完')
"""
work1
work2
work3
end work3
end work2
end work1
p1.name: Process-1
p2.name: Process-2
p3.name: Process-3
主进程执行完
Process finished with exit code 0
"""
E、继承Process重写run方法实现多进程
#使用继承方式创建进程（面向对象的方式，新建一个进程类，这个类继承Process
#重写Process中的run方法）
#导入模块
from multiprocessing import Process
from time import sleep
import time
#定义类
class ClockProcess(Process):
#重新初始化方法
def __init__(self,interval):
Process.__init__(self)
self.interval=interval
#重新run()
def run(self):
print('子进程开始执行的时间:{}'.format(time.ctime()))
sleep(self.interval)
print('子进程结束的时间：{}'.format(time.ctime()))
if __name__ == '__main__':
#创建子进程
p=ClockProcess(3)
#调用子进程
p.start()
p.join()
print('主进程执行完')
"""
子进程开始执行的时间:Sat Oct 31 16:00:51 2020
子进程结束的时间：Sat Oct 31 16:00:54 2020
主进程执行完
"""
进程池
介绍：
"""
在利用python 进行系统管理是，特别是同时操作多个文件目录，或者远程控制多台主机，
并行操作就可以节省大量的时间。单被操作对象数目不大时，可以直接利用multiprocessing
中的Process动态生成多个进程，但是当数量过多目标多时，手动去限制进程数量却又太过
繁琐，此时可以发挥进程池的作用
Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没满，
那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已达规定最大值，那么
该请求就会等待，直到池中有进程结束，才会创建新的进程。Pool的语法格式
Pool([numprocess[,initializer[,initargs]]])
其中numprocess是要创建的进程数。如果省略池参数，这将使用cpu_count()(4核cpu创建四个进程池)的值。
initializer是每个工作进程启动时要执行的可调用对象。
initarges是要传递给initializer的参数元组
initializer默认为None
实例方法：
apply(func,[,args,[,kwargs]])
在一个池工作进程中执行函数（*args,**kwargs）然后返回结果
apply_async(func[,args [,kwargs[,callback]]])
#apply_async(self, func, args=(), kwds={}, callback=None,error_callback=None):
在一个池工作进程中异步执行函数（*args,**kwargs）,然后返回结果。
此方法的结果是AsyncResult类的实例，搜会可以用于获得最终结果。
callback是可调用对象，接受输入参数。
当func的结果变为可用时，将立即传递给callback。
callback禁止执行任何阻塞操作，否则将阻塞操作接收其他异步操作的结果
close（）
关闭进程池，防止进行进一步操作。如果还有挂起的操作，它们将在工作进程终止之前完成
"""
a、进程池：非阻塞进程池的使用
Pool().apply_async(func,args=(msg,))
#非阻塞进程池的使用
from multiprocessing import Pool
import time
def func(msg):
print('start:',msg)
time.sleep(3)
print('end:',msg)
if __name__=='__main__':
#创建进程池
pool=Pool(3)
#添加任务
for i in range(1,6):
msg='任务%d'%i
pool.apply_async(func,args=(msg,))
#如果进程池不再接受新的请求，调用close
pool.close()
#等待子进程接受
pool.join() #当join被注释，则Process finished with exit code 0主进程直接结束
print('主进程结束')
"""
pool里面生成了三条process，每条process都执行，先开始三个任务，其他任务等待当前
任务结束，任务2,1结束后，4,5会进到process开始执执行故:start：任务4 start:任务5
start: 任务1
start: 任务2
start: 任务3
end: 任务2
end: 任务1
start: 任务4
start: 任务5
end: 任务3
end: 任务5
end: 任务4
主进程结束
"""
b、进程池：阻塞进程池
Pool().apply(func,(msg,))
#阻塞进程中Pool()的apply（）方法
import multiprocessing
import time
#进程执行的任务函数
def func(msg):
print('start:',msg)
time.sleep(3)
print('end:',msg)
if __name__ == '__main__':
#创建初始化3的进程池
pool=multiprocessing.Pool(3)
#添加任务
for i in range(1,6):
msg='任务%d'%i
pool.apply(func,(msg,))
#如果进程池不再接收新的请求 调用close
pool.close()
#等待子进程结束
pool.join()
"""
一个任务执行完再执行下一个任务
start: 任务1
end: 任务1
start: 任务2
end: 任务2
start: 任务3
end: 任务3
start: 任务4
end: 任务4
start: 任务5
end: 任务5
"""
c、进程池：多个进程之间数据不共享
#多个进程之间数据不共享
from multiprocessing import Process
num=10
def work1():
global num
num+=5
print('子进程1运行后：num的值',num)
def work2():
global num
num+=10
print('子进程2运行后：num的值',num)
if __name__ == '__main__':
print('主进程开始运行')
#创建子进程
p1=Process(target=work1)
p2=Process(target=work2)
#启动子进程
p1.start()
p2.start()
#主进程等待子进程结束
p1.join()
p2.join()
print('全局变量num：',num)
"""
主进程开始运行
子进程1运行后：num的值 15
子进程2运行后：num的值 20
全局变量num： 10
"""
Queue队列
a、Queue简述
""" Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递 put方法用以插入数据到队列中 put方法还有两个可选参数：blocked和timeout。 如果blocked为True(默认为True),并且timeout为正值，该方法会阻塞timeout指定的时间 如果超时，会抛出Queue.full异常。如果blocked为False，但该Queue已满，会立即抛出 Queue.full异常。 如果blocked为False，但该Queue已满，会立即抛出Queue.full异常。 get方法可以从队列读取并且删除一个元素。 同样，get方法有两个可选参数：blocked和timeout 如果blocked为True(默认为True),并且timeout为正值，那么在等待时间内没有取到任何元素 会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值 可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常 """
b、Queue的属性和方法
Queue().put()， Queue.get() Queue().full Queue().empty Queue.qsize()查看队列的大小
#Queue队列方法使用
from multiprocessing import Queue
#创建一个队列
q=Queue(3)#可以指定队列的大小，如果不写默认的队列是无限的
#向队列中插入元素
q.put('消息1')
q.put('消息2')
q.put('消息3')
#q.put('消息4',block=True,timeout=1)这里抛出queue.Full异常
#判断当前队列是否已满
print('判断当前队列是否已满：',q.full()) #判断当前队列是否已满： True
#通过判断是否已满，选择是否在队列中插入消息4
if not q.full():
q.put('消息4',block=True,timeout=1)
#读取并删除元素get
print(q.get()) #消息1
print(q.get()) #消息2
print(q.get()) #消息3
#队列为空时
#print(q.get(block=True,timeout=1))#异常_queue.Empty
#通过判断是否队列已空,选择是否读取元素
if not q.empty():
q.get(block=True,timeout=1)
#查看队列大小
print('当前队列大小为：',q.qsize()) # 当前队列大小为： 0,因为已全部get完
#通过队列长度迭代获取队列内元素
q2=Queue(4)
q2.put('数量1')
q2.put('数量2')
q2.put('数量3')
q2.put('数量4')
print('当前队列的大小：',q2.qsize())
for i in range(q2.qsize()):
print(q2.get())
#此时查看队列的大小
print('此时的队列大小：',q2.qsize())
'''
当前队列的大小： 4
数量1
数量2
数量3
数量4
此时的队列大小： 0
'''
c、Queue队列实现进程间通信
#Pool的方式创建进程，使用multiprocessing.Manager()中的Queue()来完成进程间的通信，
#而不是multiprocessing.Queue(),否则会抛出异常（RunningtimeError）
#使用进程池创建进程完成进程之间的通信
from multiprocessing import Pool,Manager
from time import sleep
def write(q):
a=['a','b','c','d']
for i in a:
print('开始写入的值：%s'%i)
q.put(i)
sleep(1)
def reader(q):
for i in range(q.qsize()):
print('读取到的值：%s'%q.get())
sleep(1)
if __name__=='__main__':
#创建队列
q=Manager().Queue()
p=Pool(4)
p.apply(write,(q,))
p.apply(reader,(q,))
p.close()
p.join()
"""
开始写入的值：a
开始写入的值：b
开始写入的值：c
开始写入的值：d
读取到的值：a
读取到的值：b
读取到的值：c
读取到的值：d
"""
线程
概述
线程
线程也是实现多任务的一种方法，一个进程中，也经常需要同时做多件事，就需要同时运行多个
‘子任务’，这些子任务就是线程。一个进程可以拥有多个并行的线程，其中每一个线程，共享
当前的资源。
区别 进程 线程
根本区别 作为资源分配的单位 调度和执行的单位
开销 每一个进程都是独立的代码和 线程可以看作是轻量级的进程，
数据空间进程之间的切换会有较大的开销 多个线程共享内存，线程切换开销小
所处环境 在操作系统中，同时运行的多个任务 在程序中多个顺序流同时执行
系统在运行的时候 线程所使用的资源是
分配内存 为每一个进程分配不同的内存区域 他所属进程的资源
包含关系 一个进程内可以拥有多个线程 线程是进程的一部分，所有线程有
时候称为是轻量级的进程
在使用上各有优缺点：
线程执行开销小，但不利于资源的管理和保护，进程正好相反
在python程序中，可以通过'_thread'和threading（推荐使用）这两个模块来处理线程。
在python3中thread模块已经废弃。可以使用threadting模块代替。但是为了兼容python3之前的
程序，在python3中将thread模块重命名为“_thread”.
_thread模块
通过调用里面的函数start_new_thread()来生成一个新的线程
语法格式：
_thread.start_new_thread(function,（参数1，参数2）,{})
其中function表示线程函数；args表示传递给线程函数的参数，他必须是个tuple类型；kwargs是可选参数
#创建线程start_new_thread
from _thread import start_new_thread
import time
def fun1():
print('开始运行fun1')
time.sleep(4)
print('fun1运行结束')
def fun2():
print('开始运行fun2')
time.sleep(2)
print('fun2运行结束')
if __name__ =='__main__':
print('开始运行')
#创建进程
start_new_thread(fun1,(),)
start_new_thread(fun2,(),)
time.sleep(7)
print('主进程结束')
#在start_new_thread中传入参数
import time
import _thread
def fun1(thead_name,delay):
print('线程名为{}'.format(thead_name))
time.sleep(delay)
print('{}结束'.format(thead_name))
def fun2(thead_name,delay):
print('线程名为{}'.format(thead_name))
time.sleep(delay)
print('{}结束'.format(thead_name))
if __name__=='__main__':
print('主线程开始')
_thread.start_new_thread(fun1,('线程1',2))
_thread.start_new_thread(fun1, ('线程2', 5))
time.sleep(8)
主线程开始 线程名为线程2 线程名为线程1 线程1结束 线程2结束
threading模块
a、threading:创建线程
使用 threading 模块创建线程通常有两种方式：1）使用 threading 模块中 Thread 类的构造器创建线程，即直接对类 threading.Thread 进行实例化，
并调用实例化对象的 start 方法创建线程；
2）继承 threading 模块中的 Thread 类创建线程类，即用 threading.Thread 派生出一个新的子类，将新建类实例化，
并调用其 start 方法创建线程。
b、threading:构造器方式
调用 threading.Thread 类的如下构造器创建线程：
threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)
group：指定该线程所属的线程组，目前该参数还未实现，为了日后扩展 ThreadGroup 类实现而保留。
target：用于 run() 方法调用的可调用对象，默认是 None，表示不需要调用任何方法。
args：是用于调用目标函数的参数元组，默认是 ()。
kwargs：是用于调用目标函数的关键字参数字典，默认是 {}。
daemon：如果 daemon 不是 None，线程将被显式的设置为守护模式，不管该线程是否是守护模式，如果是 None (默认值)，线程将继承当前线程的守护模式属性。
c、threading:常用方法、属性
1）threading.Thread 实例的方法、属性
| 方法 | 说明 |
|---|---|
| start() | 启动线程活动，它在一个线程里最多只能被调用一次。 |
| run() | 表示线程活动的方法。 |
| join(timeout=None) | 等待至线程中止。 |
| getName() | 返回线程名。 |
| setName() | 设置线程名。 |
| is_alive() | 返回线程是否是活动的。 |
| daemon | 是否为守护线程的标志。 |
| ident | 线程标识符，线程尚未开始返回 None，已启动返回非零整数。 |
2）threading 直接调用的方法
| 方法 | 说明 |
|---|---|
| active_count() | 返回当前存活的线程类 Thread 对象，返回个数等于 enumerate() 返回的列表长度。 |
| current_thread() | 返回当前对应调用者的 Thread 对象。 |
| get_ident() | 返回当前线程的线程标识符，它是一个非零的整数。 |
| enumerate() | 以列表形式返回当前所有存活的 Thread 对象。 |
| main_thread() | 返回主 Thread 对象。 |
| settrace(func) | 为所有 threading 模块开始的线程设置追踪函数。 |
| setprofile(func) | 为所有 threading 模块开始的线程设置性能测试函数。 |
| stack_size([size]) | 返回创建线程时用的堆栈大小。 |
E、threading:实例：线程创建
1）threading.Thread直接创建线程
threading,Thread().start()
#threading构造线程
#start 方法开始线程
import time
import threading
def fun1(thead_name,delay):
print('线程名为{}'.format(thead_name))
time.sleep(delay)
print('{}结束'.format(thead_name))
def fun2(thead_name,delay):
print('线程名为{}'.format(thead_name))
time.sleep(delay)
print('{}结束'.format(thead_name))
if __name__=='__main__':
t1=threading.Thread(target=fun1,args=('线程1',3))
t2=threading.Thread(target=fun2,args=('线程2',5))
#启动线程
t1.start() #无start则不执行线程
t2.start()
"""
线程名为线程1
线程名为线程2
线程1结束
线程2结束
Process finished with exit code 0
"""
2）继承threading.Thread类创建线程
重写threading.Thread中的方法run(),然后再调用方法start()就能创建线程，并运行方法run（）中的代码
#简单重写run方法构造线程
import time
import threading
class NewThread(threading.Thread):
def __init__(self,func_name,delay):
super().__init__()
self.func_name=func_name
self.delay=delay
def run(self):
print('线程名为{}'.format(self.func_name))
time.sleep(self.delay)
print('{}结束'.format(self.func_name))
if __name__ == '__main__':
n1=NewThread('func1',3)
n2=NewThread('func2',4)
n1.start()
n2.start()
"""
线程名为func1
线程名为func2
func1结束
func2结束
Process finished with exit code 0
"""
#重写run方法，执行函数参数
import time
import threading
def fun1(delay):
print('线程名为{}'.format(threading.current_thread().getName()))
time.sleep(delay)
print('{}结束'.format(threading.current_thread().getName()))
def fun2(delay):
print('线程名为{}'.format(threading.current_thread().getName()))
time.sleep(delay)
print('{}结束'.format(threading.current_thread().getName()))
class NewThread(threading.Thread):
def __init__(self,func,name,args):
super().__init__(target=func,name=name,args=args)
def run(self):
self._target(*self._args)
"""
源码里是这样写的：
def run(self):
"""Method representing the thread's activity.
You may override this method in a subclass. The standard run() method
invokes the callable object passed to the object's constructor as the
target argument, if any, with sequential and keyword arguments taken
from the args and kwargs arguments, respectively.
"""
try:
if self._target:
self._target(*self._args, **self._kwargs)
finally:
# Avoid a refcycle if the thread is running a function with
# an argument that has a member that points to the thread.
del self._target, self._args, self._kwargs
源代码有源于self._target(*self._args)的用法这里的self._target=taget；self._args = args(tread__init__方法中已注明)
super().__init__()继承父类中属性
"""
if __name__ == '__main__':
n1=NewThread(fun1,'thread_1',(3,))
n2=NewThread(fun2,'thread_2',(3,))
n1.start()
n2.start()
n1.join()
n2.join()
"""
线程名为thread_1
线程名为thread_2
thread_2结束
thread_1结束
Process finished with exit code 0
"""
#重写run方法，实现多条线程
import time
import threading
class MyThread(threading.Thread):
def __init__(self,num):
super().__init__()
self.num = num
def run(self):
print('线程名称：', threading.current_thread().getName(), '参数：', self.num, '开始时间：', time.strftime('%Y-%m-%d %H:%M:%S'))
if __name__ == '__main__':
print('主线程开始时间：',time.strftime('%Y-%m-%d %H:%M:%S'))
t1 = MyThread(3)
t2 = MyThread(2)
t3 = MyThread(1)
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()
print('主线程结束时间：', time.strftime('%Y-%m-%d %H:%M:%S'))
F、threading:线程共享全局变量
1）一个进程内所有线程共享全局变量，多线程之间额数据共享比多进程要好
#线程共享全局变量
#定义一个全局变量
import time
from threading import Thread
num=10
def test1():
global num
for i in range(3):
num+=1
print('执行函数1')
def test2():
print('执行函数2numd的值：',num)
t1=Thread(target=test1)
t2=Thread(target=test2)
t1.start()
t2.start()
t1.join()
t2.join()
"""
执行函数1
执行函数2numd的值： 13
"""
2）线程共享全局变量的弊端
可能造成多个进程同时修改一个变量(即线程非安全)，可能造成混乱。
#线程共享全局变量造成混乱
from threading import Thread
num=0
def test1():
global num
for i in range(1000000):
num+=1
print('test1中num的值：',num)
def test2():
global num
for i in range(1000000):
num+=1
print('test2中的num的值:',num)
if __name__=='__main__':
t1=Thread(target=test1)
t2=Thread(target=test2)
t1.start()
t2.start()
t1.join()
t2.join()
"""
cpu中最多只能运行一条线程，两条线程会抢占cpu资源，导致test1可能与test2执行的值有重叠
频繁修改会造成混乱
test1中num的值： 1244792
test2中的num的值: 1326689
"""
G、threading:互斥锁Lock（）
如果多线程共同对某个数据修改，则可能出现不可预料的结果，为了保证数据的正确性，需要对多个线程进行同步
最简单的同步机制就是引入互斥锁
锁有两种状态：锁定和未锁定。某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改：
直到该线程释放资源，将资源的状态变为“非锁定”状态，其他的线程才能再次锁定该资源。
互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性
使用Thread对象的lock可以实现简单的线程同步，有上锁acquire方法和释放release方法，对于那些需要每次只允许一个线程
操作的数据，可以将其操作放到acquire和release方法之间#先执行循环，循环结束再解锁
1）锁的使用
mutex = threading.Lock() #锁的使用 #创建锁 mutex = threading.Lock() #锁定 mutex.acquire([timeout]) #释放 mutex.release()
2）锁的应用
3）在循环开始后进行上锁解锁
#循环中交替上锁解锁，可能存在交替解锁上锁，导致耗时更长
#线程互斥锁
from threading import Thread,Lock
import time
num=10
#创建一把锁
lock_num=Lock()
def test1():
global num
for i in range(100000):
lock_num.acquire()
num+=1
lock_num.release()
print(num)
def test2():
global num
for i in range(100000):
lock_num.acquire()
num+=1
lock_num.release()
print(num)
if __name__=='__main__':
start_time=time.time()
t1=Thread(target=test1)
t2= Thread(target=test2)
t1.start()
t2.start()
t1.join()
t2.join()
finish_time = time.time()
print('耗时:',finish_time-start_time)
# """
# 195182
# 200010
# 耗时: 0.016956090927124023
#
4）RLock(可重入锁)
是一个可以被同一个线程请求多次的同步指令指令。RLock使用了“拥有的线程”和“递归等级”的概念，处于锁定状态时，RLock被某个线程拥有。拥有RLock的线程可以再次调用acquire()，释放锁时需要调用release()相同次数。可以认为RLock包含一个锁定池和一个初始值为0的计数器，每次成功调用 acquire()/release()，计数器将+1/-1，为0时锁处于未锁定状态。
构造方法：mylock = Threading.RLock()
实例方法：acquire([timeout])/release(): 跟Lock差不多。
#可重入锁使用
from threading import Thread,RLock
import time
num=10
#创建递归锁
lock_num=RLock()
def test1():
global num
for i in range(1000000):
lock_num.acquire()
num+=1
lock_num.release()
print(num)
def test2():
global num
for i in range(1000000):
lock_num.acquire()
num+=1
lock_num.release()
print(num)
if __name__=='__main__':
start_time=time.time()
t1=Thread(target=test1)
t2= Thread(target=test2)
t1.start()
t2.start()
t1.join()
t2.join()
finish_time = time.time()
print('耗时:',finish_time-start_time)
"""
1924026
2000010
耗时: 0.8287830352783203
#
"""
H、threading:线程同步的应用
线程同步就是协同步调，按预定的先后次序进行运行。
如进程、线程同步，可以理解为线程或进程A和B一块配合，A执行到一定程度是要依靠B的某个结果，预算停下来，示意B运行，B运行后将结果给A，A 继续运行
1）示例线程同步 多lock
#线程同步：
from threading import Thread,Lock
import time
lock1=Lock()
lock2=Lock()
lock3=Lock()
lock2.acquire()
lock3.acquire()
class Task1(Thread):
def run(self):
while True:
if lock1.acquire():
print("lock1加锁")
time.sleep(1)
lock2.release()
class Task2(Thread):
def run(self):
while True:
if lock2.acquire():
print("lock2释放")
time.sleep(1)
lock3.release()
class Task3(Thread):
def run(self):
while True:
if lock3.acquire():
print("lock3释放")
time.sleep(1)
lock1.release()
if __name__=="__main__":
t1=Task1()
t2=Task2()
t3=Task3()
t1.start()
t2.start()
t3.start()
"""
#线程同步：
from threading import Thread,Lock
import time
lock1=Lock()
lock2=Lock()
lock3=Lock()
lock2.acquire()
lock3.acquire()
class Task1(Thread):
def run(self):
while True:
if lock1.acquire():
print("lock1加锁")
time.sleep(1)
lock2.release()
class Task2(Thread):
def run(self):
while True:
if lock2.acquire():
print("lock2释放")
time.sleep(1)
lock3.release()
class Task3(Thread):
def run(self):
while True:
if lock3.acquire():
print("lock3释放")
time.sleep(1)
lock1.release()
if __name__=="__main__":
t1=Task1()
t2=Task2()
t3=Task3()
t1.start()
t2.start()
t3.start()
.....
"""
i、threading:消费者生产模式queue模块Queue方法的应用
""" Queue.put(item, block=True, timeout=None) item，放入队列中的数据元素。 block，当队列中元素个数达到上限继续往里放数据时：如果 block=False，直接引发 queue.Full 异常；如果 block=True，且 timeout=None，则一直等待直到有数据出队列后可以放入数据；如果 block=True，且 timeout=N，N 为某一正整数时，则等待 N 秒，如果队列中还没有位置放入数据就引发 queue.Full 异常。 timeout，设置超时时间。 Queue.get(block=True, timeout=None) 从队列中取出数据并返回该数据内容。 block，当队列中没有数据元素继续取数据时：如果 block=False，直接引发 queue.Empty 异常；如果 block=True，且 timeout=None，则一直等待直到有数据入队列后可以取出数据；如果 block=True，且 timeout=N，N 为某一正整数时，则等待 N 秒，如果队列中还没有数据放入的话就引发 queue.Empty 异常。 timeout，设置超时时间。 """
1)生产消费模式while循环
from queue import Queue
import threading
import time
class Product(threading.Thread):
global queue
def run(self):
while True:
if queue.qsize()<1000:
for i in range(1,21):
cont='生产第'+str(i)+'个'
print(cont)
queue.put(cont)
time.sleep(1)
class Consumer(threading.Thread):
global queue
def run(self):
while True:#保证循环生产循环消费
if queue.qsize()>10:
for i in range(1,3): #这里从队列里面取数数消费
cont=queue.get()
print('消费了'+cont)
time.sleep(2)
if __name__=='__main__':
queue=Queue()
p=Product()
C=Consumer()
p.start()
C.start()
生产第1个 生产第2个 生产第3个 生产第4个 生产第5个 生产第6个 生产第7个 生产第8个 生产第9个 生产第10个 生产第11个 生产第12个 生产第13个 生产第14个 生产第15个 消费了生产第1个 消费了生产第2个 生产第16个
2）生产消费模式不循环
在队列中放置一个特殊的值，当消费者读到这个值的时候，终止执行
from queue import Queue
from threading import Thread
# 用来表示终止的特殊对象
_sentinel = object()
# A thread that produces data
def producer(out_q):
for i in range(10):
print("生产")
out_q.put(i)
out_q.put(_sentinel)
# A thread that consumes data
def consumer(in_q):
while True:
data = in_q.get()
if data is _sentinel:
in_q.put(_sentinel)
break
else:
print("消费", data)
q=Queue()
t1=Thread(target=producer,args=(q,))
t2=Thread(target=consumer,args=(q,))
t1.start()
t2.start()
生产 生产 生产 生产 生产 生产 生产 消费 0 消费 生产 生产 生产 1 消费 2 消费 3 消费 4 消费 5 消费 6 消费 7 消费 8 消费 9 Process finished with exit code 0
J、threading:threading.local()
我们知道多线程环境下，每一个线程均可使用所属进程的全局变量。如果一个线程对全局变量进行了修改，将会影响到其他所有的线程对全局变量的计算操作，
从而出现数据混乱，即为脏数据。为了避免多个线程同时对变量进行修改，引入了线程同步机制，通过互斥锁来控制对全局变量的访问。所以有时候线程使用局部变量
比全局变量好，因为局部变量只有线程自身可以访问，同一个进程下的其他线程不可访问
局部变量作为参数传递函数调用时传递起来很麻烦，层层调用函数都需要传递同一个参数，如果使用全局变量也不能解决
因此python还提供了ThreadLocal变量，它本身是一个全局变量，但是每个线程却可以利用它来保存属于自己的私有数据，这些是有数据对其他线程也是不可见的
1）threading.local()的使用
1、创建全局ThreadLocal对象
2、threading.locall绑定为函数1内的某一参数
3. 函数2传入参数与函数1中的threading.local赋值
4. 在函数2内调用函数1，加入线程内实施
#TreadLocal局部参数的使用
import threading
local=threading.local()
def get_local():
name=local.name
print('当前线程名为%s,学生姓名为%s'%(threading.current_thread().name,name))
def tast1(stu_name):
local.name=stu_name
get_local()
t1=threading.Thread(target=tast1,args=('张三',),name='线程1')
t2=threading.Thread(target=tast1,args=('李四',),name='线程2')
t1.start()
t2.start()
"""
当前线程名为线程1,学生姓名为张三
当前线程名为线程2,学生姓名为李四
"""
