1.概念 进程 线程
# 进程 就是一个程序在一个数据集上的一个动态执行过程 是最小的资源单元 # 进程是由 程序 数据集 进程控制块组成 本质上是一段程序的运行过程 # 1. 一个程序至少有一个进程,一个进程至少有一个线程.(进程可以理解成线程的容器) # # 2. 进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的运行效率。 # # 3. 线程在执行过程中与进程还是有区别的。每个独立的线程有一个程序运行的入口、顺序执行序列和 # 程序的出口。但是线程不能够独立执行，必须依存在应用程序中，由应用程序提供多个线程执行控制。 # # 4. 进程是具有一定独立功能的程序关于某个数据集合上的一次运行活动,进程是系统进行资源分配和调度的一个独立单位. # 线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程 # 自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈)但是 # 它可与同属一个进程的其他的线程共享进程所拥有的全部资源. # 一个线程可以创建和撤销另一个线程;同一个进程中的多个线程之间可以并发执行. # 线程 ：降低上下文切换的消耗 提高系统的并发性 是最小执行单元
# python GIL # 无论你启多少个线程，你有多少个cpu, Python在执行的时候会淡定的在同一时刻只允许一个线程运行 只支持单核
2.调用多线程2种方法
A.直接调用
import threading import time def sayhi(num): #定义每个线程要运行的函数 print("running on number:%s" %num) time.sleep(3) if __name__ == '__main__': t1 = threading.Thread(target=sayhi,args=(1,)) #生成一个线程实例 t2 = threading.Thread(target=sayhi,args=(2,)) #生成另一个线程实例 t1.start() #启动线程 t2.start() #启动另一个线程 print(t1.getName()) #获取线程名 print(t2.getName())
B.继承式调用
import threading import time class MyThread(threading.Thread): def __init__(self,num): threading.Thread.__init__(self) self.num = num def run(self):#定义每个线程要运行的函数 print("running on number:%s" %self.num) time.sleep(3) if __name__ == '__main__': t1 = MyThread(1) t2 = MyThread(2) t1.start() t2.start() print("ending......")
3.join&Daemon方法
参考： https://www.cnblogs.com/cnkai/p/7504980.html 1.join Python多线程与多进程中join()方法的效果是相同的 join() 没执行完 下面代码不执行。 2.setDaemon 设置守护线程 下面仅以多线程为例： 首先需要明确几个概念： 知识点一： 当一个进程启动之后，会默认产生一个主线程，因为线程是程序执行流的最小单元，当设置多线程时，主线程会创建多个子线程， 在python中，默认情况下（其实就是setDaemon(False)），主线程执行完自己的任务以后，就退出了，此时子线程会继续执行自己的任务， 直到自己的任务结束，例子见下面一。 知识点二： 当我们使用setDaemon(True)方法，设置子线程为守护线程时，主线程一旦执行结束，则全部线程全部被终止执行，可能出现的情况就是， 子线程的任务还没有完全执行结束，就被迫停止，例子见下面二。 知识点三： 此时join的作用就凸显出来了，join所完成的工作就是线程同步，即主线程任务结束之后，进入阻塞状态，一直等待其他的子线程执行结束之后 ，主线程在终止，例子见下面三。 知识点四： join有一个timeout参数： 当设置守护线程时，含义是主线程对于子线程等待timeout的时间将会杀死该子线程，最后退出程序。所以说，如果有10个子线程， 全部的等待时间就是每个timeout的累加和。简单的来说，就是给每个子线程一个timeout的时间，让他去执行，时间一到，不管任务 有没有完成，直接杀死。 没有设置守护线程时，主线程将会等待timeout的累加和这样的一段时间，时间一到，主线程结束，但是并没有杀死子线程， 子线程依然可以继续执行，直到子线程全部结束，程序退出。 ''' # 例一 # setDaemon(False) import threading import time def run(): time.sleep(2) print('当前线程的名字是： ', threading.current_thread().name) time.sleep(2) if __name__ == '__main__': start_time = time.time() print('这是主线程：', threading.current_thread().name) thread_list = [] for i in range(5): t = threading.Thread(target=run) thread_list.append(t) for t in thread_list: t.start() print('主线程结束！', threading.current_thread().name) print('一共用时：', time.time() - start_time) ''' 我们的计时是对主线程计时，主线程结束，计时随之结束，打印出主线程的用时。 主线程的任务完成之后，主线程随之结束，子线程继续执行自己的任务，直到全部的子线程的任务全部结束，程序结束。 ''' print("--------------------------------------------------") # 例二 # setDaemon(True) import threading import time def run(): time.sleep(2) print('当前线程的名字是： ', threading.current_thread().name) time.sleep(2) if __name__ == '__main__': start_time = time.time() print('这是主线程：', threading.current_thread().name) thread_list = [] for i in range(5): t = threading.Thread(target=run) thread_list.append(t) for t in thread_list: t.setDaemon(True) t.start() print('主线程结束了！', threading.current_thread().name) print('一共用时：', time.time() - start_time) # 主线程结束以后，子线程还没有来得及执行，整个程序就退出了 print("-------------------------------------------------------")
# 例三 # 类方法创建多线程 # 使用类方式需要写一个类，继承自threading.Thread类，然后重写run()方法。 import threading import time def run(): time.sleep(2) print('当前线程的名字是： ', threading.current_thread().name) time.sleep(2) if __name__ == '__main__': start_time = time.time() print('这是主线程：', threading.current_thread().name) thread_list = [] for i in range(1): t = threading.Thread(target=run) thread_list.append(t) for t in thread_list: t.setDaemon(True) t.start() for t in thread_list: t.join() print('主线程结束了！', threading.current_thread().name) print('一共用时：', time.time() - start_time) # 线程一直等待全部的子线程结束之后，主线程自身才结束，程序退出。 import threading import time class MyThread(threading.Thread): def __init__(self, sec): super(MyThread, self).__init__() self.sec = sec def run(self): print('%s 线程开始了！' % threading.current_thread().name) time.sleep(self.sec) print('%s 线程结束了！' % threading.current_thread().name) if __name__ == '__main__': print('主线程开始执行', threading.current_thread().name) s_time = time.time() my_thread_list = [] for i in range(5): my_thread = MyThread(i) my_thread_list.append(my_thread) for i in my_thread_list: i.start() for i in my_thread_list: i.join() print('一共用时：', time.time() - s_time) print('主线程结束执行', threading.current_thread().name)
join()：在子线程完成运行之前，这个子线程的父线程将一直被阻塞。
setDaemon(True)：
将线程声明为守护线程，必须在start() 方法调用之前设置， 如果不设置为守护线程程序会被无限挂起。这个方法基本和join是相反的。
当我们 在程序运行中，执行一个主线程，如果主线程又创建一个子线程，主线程和子线程 就分兵两路，分别运行，那么当主线程完成
想退出时，会检验子线程是否完成。如 果子线程未完成，则主线程会等待子线程完成后再退出。但是有时候我们需要的是 只要主线程
完成了，不管子线程是否完成，都要和主线程一起退出，这时就可以 用setDaemon方法
其他方法：
# run(): 线程被cpu调度后自动执行线程对象的run方法 # start():启动线程活动。 # isAlive(): 返回线程是否活动的。 # getName(): 返回线程名。 # setName(): 设置线程名。 threading模块提供的一些方法： # threading.currentThread(): 返回当前的线程变量。 # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
4.GIL 同步锁 死锁 递归锁
并行： 是指系统具有处理多个任务（动作）的能力
并发： 是指系统具有 同时 处理多个任务（动作）的能力
同步： 当进程执行到一个IO（等待外部数据）的时候，需要等待接受数据完成后 再执行其他动作
异步： 当进程执行到一个IO（等待外部数据）的时候，不需要等待接受数据完成后 可去执行其他动作 当接受数据的完成后 再回来执行
# 全局解释器锁 # 无论你启多少个线程，你有多少个cpu, Python在执行的时候会淡定的在同一时刻只允许一个线程运行 只支持单核
#IO密集型任务: 推荐多进程 + 携程
#计算密集任务：用多进程，不推荐用多线程
# 同步锁 Lock() acquire() R.release() 让代码串行运行 # 多个线程都在同时操作同一个共享资源，所以造成了资源破坏，怎么办呢？(join会造成串行，失去所线程的意义) # 我们可以通过同步锁来解决这种问题 import time import threading def test(): global num Rlock.acquire() temp = num time.sleep(0.00001) num = temp -1 Rlock.release() # num -= num # 0 num = 100 l = [] Rlock = threading.Lock() for i in range(100): t = threading.Thread(target=test) l.append(t) for i in l: i.start() for i in l: i.join() print(num)
# 递归锁 死锁 # 在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁， # 因为系统判断这部分资源都正在使用，所有这两个线程在无外力作用下将一直等待下去。下面是一个死锁的例子： # 为了支持在同一线程中多次请求同一资源，python提供了“可重入锁”：threading.RLock。RLock内部维护着一个Lock和一个counter变量， # counter记录了acquire的次数，从而使得资源可以被多次acquire。直到一个线程所有的acquire都被release，其他的线程才能获得资源。 import threading, time class myThread(threading.Thread): # def doA(self): # lockA.acquire() # print(self.name, "gotlockA", time.ctime()) # time.sleep(1) # # lockB.acquire() # print(self.name, "gotlockB", time.ctime()) # lockB.release() # # lockA.release() # # def doB(self): # lockB.acquire() # print(self.name, "gotlockB", time.ctime()) # time.sleep(2) # # lockA.acquire() # print(self.name, "gotlockA", time.ctime()) # lockA.release() # # lockB.release() def doA(self): r_lock.acquire() print(self.name, "gotlockA", time.ctime()) time.sleep(1) r_lock.acquire() print(self.name, "gotlockB", time.ctime()) r_lock.release() r_lock.release() def doB(self): r_lock.acquire() print(self.name, "gotlockB", time.ctime()) time.sleep(2) r_lock.acquire() print(self.name, "gotlockA", time.ctime()) r_lock.release() r_lock.release() def run(self): self.doA() self.doB() if __name__ == "__main__": # lockA = threading.Lock() # lockB = threading.Lock() r_lock = threading.RLock() threads = [] for i in range(5): threads.append(myThread()) for t in threads: t.start() for t in threads: t.join() # 等待线程结束，后面再讲。
# 同步条件
event = threading.Event() # a client thread can wait for the flag to be set 等待set的状态 如果状态默认值是False等待 不执行下面语句， 是True时就执行下面语句 event.wait() # a server thread can set or reset it event.set() event.clear()
# 信号量用来控制线程并发数的，BoundedSemaphore或Semaphore管理一个内置的计数 器，每当调用acquire()时-1，调用release()时+1。 # 计数器不能小于0，当计数器为 0时，acquire()将阻塞线程至同步锁定状态，直到其他线程调用release()。(类似于停车位的概念) # BoundedSemaphore与Semaphore的唯一区别在于前者将在调用release()时检查计数 器的值是否超过了计数器的初始值，如果超过了将抛出一个异常。 import threading,time class myThread(threading.Thread): def run(self): if semaphore.acquire(): print(self.name) time.sleep(5) semaphore.release() if __name__=="__main__": semaphore=threading.Semaphore(5) thrs=[] for i in range(100): thrs.append(myThread()) for t in thrs: t.start()
5. 队列(queue) 多线程
涉及数据安全用队列
列表是不安全的数据结构 单线程中可用列表
#!/usr/bin/python env # coding:utf-8 # 对象三种模式 ''' 三种模式是 1.默认是先进先出 FIFO First In First Out 2.LifoQueue 后进先出 FILO First In Least Out 3.PriorityQueue() 优先级 '''
import queue
q = queue.LifoQueue() #默认先进先出,后进先出
q.put(12)
q.put("hello world")
q.put({"name":"yuan"})
print(q.empty())
print(q.full())
print(q.qsize())
print(q.task_done())
for i in range(q.qsize()):
data = q.get()
print(data)
print("---------------------------")
print(q.empty())
print(q.full())
print(q.task_done())
创建一个“队列”对象 import Queue q = Queue.Queue(maxsize = 10) Queue.Queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过Queue的构造函数的可选参数maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。 将一个值放入队列中 q.put(10) 调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数，默认为 1。如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0，put方法将引发Full异常。 将一个值从队列中取出 q.get() 调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True， get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。 Python Queue模块有三种队列及构造函数: 1、Python Queue模块的FIFO队列先进先出。 class queue.Queue(maxsize) 2、LIFO类似于堆，即先进后出。 class queue.LifoQueue(maxsize) 3、还有一种是优先级队列级别越低越先出来。 class queue.PriorityQueue(maxsize) 此包中的常用方法(q = Queue.Queue()): q.qsize() 返回队列的大小 q.empty() 如果队列为空，返回True,反之False q.full() 如果队列满了，返回True,反之False q.full 与 maxsize 大小对应 q.get([block[, timeout]]) 获取队列，timeout等待时间 q.get_nowait() 相当q.get(False) 非阻塞 q.put(item) 写入队列，timeout等待时间 q.put_nowait(item) 相当q.put(item, False) q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号 q.join() 实际上意味着等到队列为空，再执行别的操作
6.生产者 消费者
生产者消费者模型：
为什么要使用生产者和消费者模式
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
这就像，在餐厅，厨师做好菜，不需要直接和客户交流，而是交给前台，而客户去饭菜也不需要不找厨师，直接去前台领取即可，这也是一个结耦的过程。
import time,random import queue,threading q = queue.Queue() def Producer(name): count = 0 while count <10: print("making........") time.sleep(random.randrange(3)) q.put(count) print('Producer %s has produced %s baozi..' %(name, count)) count +=1 #q.task_done() #q.join() print("ok......")
def Consumer(name): count = 0 while count <10: time.sleep(random.randrange(4)) if not q.empty(): data = q.get() #q.task_done() #q.join() print(data) print(' 33[32;1mConsumer %s has eat %s baozi... 33[0m' %(name, data)) else: print("-----no baozi anymore----") count +=1 p1 = threading.Thread(target=Producer, args=('A',)) c1 = threading.Thread(target=Consumer, args=('B',)) # c2 = threading.Thread(target=Consumer, args=('C',)) # c3 = threading.Thread(target=Consumer, args=('D',)) p1.start() c1.start() # c2.start() # c3.start()
7.多进程
A.创建进程方法
#!/usr/bin/python env # coding:utf-8 # 1.新建多进程 # import time # from multiprocessing import Process # # # def multf(name): # time.sleep(2) # print('hello, ', name, time.ctime()) # # # if __name__ == '__main__': # l = list() # for i in range(3): # p = Process(target=multf, args=("wangwu",)) # l.append(p) # p.start() # # for i in l: # i.join() # # print("end") # 2.通过类创建多进程 import time from multiprocessing import Process import os class MyMulitProcess(Process): def __init__(self): super(MyMulitProcess, self).__init__() #self.name = name def run(self): time.sleep(1) print('hello, ', self.name, time.ctime()) def multf(name): time.sleep(2) print('hello, ', name, time.ctime()) def info(title): print("title: ", title) print("parent process: ", os.getppid()) # 这是pycharm程序的进程号 print("process id: ", os.getpid()) def f(name): info("function f") print("hello, ", name) if __name__ == '__main__': # l = list() # for i in range(3): # p = MyMulitProcess() # p.daemon = "True" # 设置守护进程 # p.start() # l.append(p) # # for i in l: # i.join() # # print("end") l = [] info("main process line") time.sleep(1) print("----------------------------") p_info = Process(target=info, args=("info",)) p_f = Process(target=f, args=("f",)) l.append(p_info) l.append(p_f) for i in l: i.start() for i in l: i.join print("--------------")
B.Process类
构造方法：
Process([group [, target [, name [, args [, kwargs]]]]])
group: 线程组，目前还没有实现，库引用中提示必须是None；
target: 要执行的方法；
name: 进程名；
args/kwargs: 要传入方法的参数。
实例方法：
is_alive()：返回进程是否在运行。
join([timeout])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。
start()：进程准备就绪，等待CPU调度
run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。
terminate()：不管任务是否完成，立即停止工作进程
属性：
daemon：和线程的setDeamon功能一样
name：进程名字。
pid：进程号。
8.进程队列 管道 数据共享 进程同步 进程池
# # 进程队列 Queue from multiprocessing import Process, Queue import queue def f(q,n): #q.put([123, 456, 'hello']) q.put(n*n+1) print("son process",id(q)) # son process 7349552 if __name__ == '__main__': q = Queue() #try: q=queue.Queue() print("main process",id(q)) # main process 29347952 for i in range(3): p = Process(target=f, args=(q,i)) p.start() print(q.get()) # 1 print(q.get()) # 2 print(q.get()) # 5 # 管道 Pipe from multiprocessing import Pipe, Process def f(conn): conn.send("i am f") response = conn.recv() print("child_conn recv: ", response) # child_conn recv: hello son conn.close() print("q_ID2: ", id(conn)) # q_ID2: 29159312 if __name__ == '__main__': parent_conn, child_conn = Pipe() # 双向通道 print("q_ID1: ", id(child_conn)) # q_ID1: 17134512 p = Process(target=f, args=(child_conn,)) p.start() print("parent_conn recv: ", parent_conn.recv()) # parent_conn recv: i am f parent_conn.send("hello son") p.join() # 数据共享 Manager # Queue和pipe只是实现了数据交互，并没实现数据共享，即一个进程去更改另一个进程的数据。 from multiprocessing import Process, Manager def f(d, l, n): d[n] = '1' # {0:"1"} d['name'] = 'zhangsan' # {'name': 'zhangsan'} l.append(n) # {0: '1','name': 'zhangsan'} print("son process: ", id(d), id(l)) if __name__ == '__main__': with Manager() as manager: d = manager.dict() l = manager.list(range(5)) print("Main Process: ", id(d), id(l)) '''Main Process: 21815184 21897360 son process: 9672176 9673456 son process: 21730800 21732080 son process: 19502576 19503856 son process: 33134096 33135376 son process: 19830256 19831536''' p_list = [] for i in range(5): p = Process(target=f, args=(d, l, i)) p.start() p_list.append(p) for res in p_list: res.join() print(d) # {0: '1', 1: '1', 2: '1', 3: '1', 4: '1', 'name': 'zhangsan'} print(l) # [0, 1, 2, 3, 4, 0, 1, 3, 2, 4] # 进程同步 Lock from multiprocessing import Process, Lock import time def f(l, i): l.acquire() # time.sleep(1) # 串行 print('hello world %s' % i) l.release() if __name__ == '__main__': lock = Lock() for num in range(10): Process(target=f, args=(lock, num)).start()
# 进程池
# 进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。
# 进程池中有两个方法：
# apply 同步 阻塞
# apply_async 异步
from multiprocessing import Process, Pool
import time
import os
def Foo(num):
time.sleep(2)
print("Foo pid : ", os.getpid())
t = time.strftime("%Y-%m-%d %X")
return "%s Foo %s" % (str(t),num)
def Log(args):
print("Log pid : ", os.getpid())
print(args)
print("-----------------------------------")
if __name__ == '__main__':
pool = Pool(5)
print("Main pid : ", os.getpid())
print("-----------------------------------")
for i in range(1, 100):
# 同步
# pool.apply_async(func=Foo, args=(i,)) #异步
# pool.apply(func=Foo, args=(i,)) #同步
# 异步
# callback回调函数:某个动作执行后或者函数执行后，再执行这个函数
# 将Foo的返回值返回给Bar Bar在主进程下调用
pool.apply_async(func=Foo, args=(i,), callback=Log)
pool.close()
pool.join()
print("ending.....")
参考：
http://www.cnblogs.com/yuanchenqi/articles/6248025.html