1.多线程实现多任务
我们前面编写的所有的Python程序,都是执行单任务的进程,也就是只有一个线程。如果我们要同时执行多个任务怎么办?
有两种解决方案:
- 一种是启动多个进程,每个进程虽然只有一个线程,但多个进程可以一块执行多个任务。
- 还有一种方法是启动一个进程,在一个进程内启动多个线程,这样,多个线程也可以一块执行多个任务。
当然还有第三种方法,就是启动多个进程,每个进程再启动多个线程,这样同时执行的任务就更多了,当然这种模型更复杂,实际很少采用。
总结一下就是,多任务的实现有3种方式:
- 多进程模式;
- 多线程模式;
- 多进程+多线程模式。
import threading, time def dance(): for i in range(50): time.sleep(0.2) print('我正在跳舞') def sing(): for i in range(50): time.sleep(0.2) print('我正在唱歌') # 多个任务同时执行 # Python里执行多任务: 多线程、多进程、多进程+多线程 # dance() # singe() # target 需要的是一个函数,用来指定线程需要执行的任务 t1 = threading.Thread(target=dance) # 创建了线程1 t2 = threading.Thread(target=sing) # 创建了线程2 # 启动线程 t1.start() t2.start()
注意:
- 并发:指的是任务数多于cpu核数,通过操作系统的各种任务调度算法,实现用多个任务“一起”执行(实际上总有一些任务不在执行,因为切换任务的速度相当快,看上去一起执行而已)
- 并行:指的是任务数小于等于cpu核数,即任务真的是一起执行的。
2.线程访问全局变量
在一个进程内的所有线程共享全局变量,很方便在多个线程间共享数据。缺点就是,线程是对全局变量随意修改可能造成多线程之间对全局变量的混乱(即线程非安全)。
import threading import time # 多个线程可以同时操作一个全局变量(多个线程共享全局变量) # 线程安全问题 ticket = 20 def sell_ticket(): global ticket while True: # ticket = 1 线程1:1 线程2: 1 if ticket > 0: time.sleep(1) # 线程1: ticket=1 线程2:ticket=1 ticket -= 1 # 线程1: ticket = 0 线程2:ticket=-1 print('{}卖出一张票,还剩{}张'.format(threading.current_thread().name, ticket)) else: print('票卖完了') break t1 = threading.Thread(target=sell_ticket, name='线程1') t2 = threading.Thread(target=sell_ticket, name='线程2') t1.start() t2.start()
3.线程锁的使用
3.1同步
当多个线程几乎同时修改某一个共享数据的时候,需要进行同步控制。同步就是协同步调,按预定的先后次序进行运行。线程同步能够保证多个线程安全访问竞争资源,最简单的同步机制是引入互斥锁。
3.2互斥锁
互斥锁为资源引入一个状态:锁定/非锁定
某个线程要更改共享数据时,先将其锁定,此时资源的状态为“锁定”,其他线程不能更改;直到该线程释放资源,将资源的状态变成“非锁定”,其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作,从而保证了多线程情况下数据的正确性。
threading模块中定义了Lock类,可以方便的处理锁定:
# 创建锁 mutex = threading.Lock() # 锁定 mutex.acquire() # 释放 mutex.release()
注意:
- 如果这个锁之前是没有上锁的,那么acquire不会堵塞
- 如果在调用acquire对这个锁上锁之前 它已经被 其他线程上了锁,那么此时acquire会堵塞,直到这个锁被解锁为止。
- 和文件操作一样,Lock也可以使用with语句快速的实现打开和关闭操作。
3.3使用互斥锁解决卖票问题
import threading import time ticket = 20 # 创建一把锁 lock = threading.Lock() def sell_ticket(): global ticket while True: print('呵呵呵') print('哈哈哈') print('ddd') print('ppp') print('sss') print('ttt') print('xxx') lock.acquire() # 加同步锁 if ticket > 0: time.sleep(1) ticket -= 1 lock.release() print('{}卖出一张票,还剩{}张'.format(threading.current_thread().name, ticket)) else: lock.release() print('票卖完了') break t1 = threading.Thread(target=sell_ticket, name='线程1') t2 = threading.Thread(target=sell_ticket, name='线程2') t1.start() t2.start()
3.4上锁过程
当一个线程调用锁的acquire()方法获得锁时,锁就进入“locked”状态。
每次只有一个线程可以获得锁。如果此时另一个线程试图获得这个锁,该线程就会变为“blocked”状态,称为“阻塞”,直到拥有锁的线程调用锁的release()方法释放锁之后,锁进入“unlocked”状态。
线程调度程序从处于同步阻塞状态的线程中选择一个来获得锁,并使得该线程进入运行(running)状态。
3.5总结
- 锁的好处:
- 确保了某段关键代码只能由一个线程从头到尾完整地执行.
- 锁的坏处:
- 阻止了多线程并发执行,包含锁的某段代码实际上只能以单线程模式执行,效率就大大地下降了。
- 由于可以存在多个锁,不同的线程持有不同的锁,并试图获取对方持有的锁时,可能会造成死锁。
4.线程间通信
线程之间有时需要通信,操作系统提供了很多机制来实现线程间的通信,其中我们使用最多的是队列Queue。经典案例:生产者和消费者。
4.1Queue的原理
Queue是一个先进先出(First In First Out)的队列,主线程中创建一个Queue对象,并作为参数传入子线程,两者之间通过put( )放入数据,通过get( )取出数据,执行了get( )函数之后队列中的数据会被同时删除。
import threading, queue import time def produce(): for i in range(10): time.sleep(0.5) print('生产++++++面包{} {}'.format(threading.current_thread().name, i)) q.put('{}{}'.format(threading.current_thread().name, i)) def consumer(): while True: time.sleep(1) # q.get()方法时一个阻塞的方法 print('{}买到------面包{}'.format(threading.current_thread().name, q.get())) q = queue.Queue() # 创建一个q # 一条生产线 pa = threading.Thread(target=produce, name='pa') pb = threading.Thread(target=produce, name='pb') pc = threading.Thread(target=produce, name='pc') # 一条消费线 ca = threading.Thread(target=consumer, name='ca') cb = threading.Thread(target=consumer, name='cb') cc = threading.Thread(target=consumer, name='cc') pa.start() pb.start() pc.start() ca.start() cb.start() cc.start()
5.多进程
5.1进程
程序:例如xxx.py这是程序,是一个静态的。
进程:一个程序运行起来后,代码+用到的资源称之为进程,它是操作系统分配资源的基本单元。
不仅可以通过线程完成多任务,进程也是可以的。
5.2进程的状态
工作中,任务数往往大于cpu的核数,即一定有一些任务正在执行,而另外一些任务在等待cpu进行执行,因此导致了有了不同的状态。
- 就绪态:运行的条件都已经满足,正在等在cpu执行。
- 执行态:cpu正在执行其功能。
- 等待态:等待某些条件满足,例如一个程序sleep了,此时就处于等待态。
5.3创建进程
multiprocessing模块就是跨平台版本的多进程模块,提供了一个Process类来代表一个进程对象,这个对象可以理解为是一个独立的进程,可以执行另外的事情。
创建子进程时,只需要传入一个执行函数和函数的参数,创建一个Process实例,用start()方法启动。
import multiprocessing, time, os def dance(n): for i in range(n): time.sleep(0.5) print('正在跳舞{},pid={}'.format(i, os.getpid())) def sing(m): for i in range(m): time.sleep(0.5) print('正在唱歌{},pid={}'.format(i, os.getpid())) if __name__ == '__main__': print('主进程的pid={}'.format(os.getpid())) # 创建了两个进程 # target 用来表示执行的任务 # args 用来传参,类型是一个元组 p1 = multiprocessing.Process(target=dance, args=(100,)) p2 = multiprocessing.Process(target=sing, args=(100,)) p1.start() p2.start()
方法说明
Process( target [, name [, args [, kwargs]]])
- target:如果传递了函数的引用,任务这个子进程就执行这里的代码
- args:给target指定的函数传递的参数,以元组的方式传递
- kwargs:给target指定的函数传递命名参数
- name:给进程设定一个名字,可以不设定
Process创建的实例对象的常用方法:
- start():启动子进程实例(创建子进程)
- is_alive():判断进程子进程是否还在活着
- join([timeout]):是否等待子进程执行结束,或等待多少秒
- terminate():不管任务是否完成,立即终止子进程
Process创建的实例对象的常用属性:
- name:当前进程的别名,默认为Process-N,N为从1开始递增的整数
- pid:当前进程的pid(进程号)
6.进程和线程的区别
6.1线程和进程
功能
- 进程,能够完成多任务,比如在一台电脑上能够同时运行多个QQ。
- 线程,能够完成多任务,比如一个QQ中的多个聊天窗口。
定义的不同
- 进程是系统进行资源分配和调度的一个独立单位。
- 线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈),但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源。
区别
- 一个程序至少有一个进程,一个进程至少有一个线程。
- 线程的划分尺度小于进程(资源比进程少),使得多线程程序的并发性高。
- 进程在执行过程中拥有独立的内存单元,而多个线程共享内存,从而极大地提高了程序的运行效率。
- 线程不能够独立执行,必须依存在进程中。
- 可以将进程理解为工厂中的一条流水线,而其中的线程就是这个流水线上的工人
优缺点
线程和进程在使用上各有优缺点:线程执行开销小,但不利于资源的管理和保护;而进程正相反。
7.多进程不能共享全局变量
from multiprocessing import Process import os nums = [11, 22] def work1(): """子进程要执行的代码""" print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums)) for i in range(3): nums.append(i) print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums)) def work2(): """子进程要执行的代码""" nums.pop() print("in process2 pid=%d ,nums=%s" % (os.getpid(), nums)) if __name__ == '__main__': p1 = Process(target=work1) p1.start() p1.join() p2 = Process(target=work2) p2.start() print('in process0 pid={} ,nums={}'.format(os.getpid(), nums))
运行结果:
in process1 pid=24240 ,nums=[11, 22] in process1 pid=24240 ,nums=[11, 22, 0] in process1 pid=24240 ,nums=[11, 22, 0, 1] in process1 pid=24240 ,nums=[11, 22, 0, 1, 2] in process0 pid=9604 ,nums=[11, 22] in process2 pid=25280 ,nums=[11]
8.进程间通信
from multiprocessing import Queue q = Queue(3) # 初始化一个Queue对象,最多可接收三条put消息 q.put("消息1") q.put("消息2") print(q.full()) # False q.put("消息3") print(q.full()) # True # 因为消息列队已满下面的try都会抛出异常,第一个try会等待2秒后再抛出异常,第二个Try会立刻抛出异常 try: q.put("消息4", True, 2) except: print("消息列队已满,现有消息数量:%s" % q.qsize()) try: q.put_nowait("消息4") except: print("消息列队已满,现有消息数量:%s" % q.qsize()) # 推荐的方式,先判断消息列队是否已满,再写入 if not q.full(): q.put_nowait("消息4") # 读取消息时,先判断消息列队是否为空,再读取 if not q.empty(): for i in range(q.qsize()): print(q.get_nowait())
说明:
初始化Queue()对象时(例如:q=Queue()),若括号中没有指定最大可接收的消息数量,或数量为负值,那么就代表可接受的消息数量没有上限(直到内存的尽头);
- Queue.qsize():返回当前队列包含的消息数量;
- Queue.empty():如果队列为空,返回True,反之False ;
- Queue.full():如果队列满了,返回True,反之False;
- Queue.get([block[, timeout]]):获取队列中的一条消息,然后将其从列队中移除,block默认值为True;
- 1)如果block使用默认值,且没有设置timeout(单位秒),消息列队如果为空,此时程序将被阻塞(停在读取状态),直到从消息列队读到消息为止,如果设置了timeout,则会等待timeout秒,若还没读取到任何消息,则抛出"Queue.Empty"异常
- 2)如果block值为False,消息列队如果为空,则会立刻抛出"Queue.Empty"异常;
- Queue.get_nowait():相当Queue.get(False);
- Queue.put(item,[block[, timeout]]):将item消息写入队列,block默认值为True;
- 1)如果block使用默认值,且没有设置timeout(单位秒),消息列队如果已经没有空间可写入,此时程序将被阻塞(停在写入状态),直到从消息列队腾出空间为止,如果设置了timeout,则会等待timeout秒,若还没空间,则抛出"Queue.Full"异常;
- 2)如果block值为False,消息列队如果没有空间可写入,则会立刻抛出"Queue.Full"异常;
- Queue.put_nowait(item):相当Queue.put(item, False);
import os, multiprocessing, time def producer(x): for i in range(10): time.sleep(0.5) print('生产了+++++++pid{} {}'.format(os.getpid(), i)) x.put('pid{} {}'.format(os.getpid(), i)) def consumer(x): for i in range(10): time.sleep(0.3) print('消费了-------{}'.format(x.get())) if __name__ == '__main__': q = multiprocessing.Queue() p1 = multiprocessing.Process(target=producer, args=(q,)) p2 = multiprocessing.Process(target=producer, args=(q,)) p3 = multiprocessing.Process(target=producer, args=(q,)) p1.start() p2.start() p3.start() c2 = multiprocessing.Process(target=consumer, args=(q,)) c2.start()
9.join方法的使用
# join 线程和进程都有join方法 import threading import time x = 10 def test(a, b): time.sleep(1) global x x = a + b t = threading.Thread(target=test, args=(1, 1)) t.start() t.join() # 让主线程等待 print(x) # 10