线程（thread）
线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的就是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并执行不同的任务。
threading
用于提供线程相关的操作。
#!/usr/bin/env python3
import threading
import time
def show(arg):
time.sleep(1)
print(‘thread‘+str(arg))
for i in range(10):
t = threading.Thread(target=show, args=(i,))
t.start()
print(‘main thread stop‘)
上述代码创建了10个前台线程，然后控制器交给了CPU，CPU根据指定算法进行调度，分片执行指令
其他方法：
- start 线程准备就绪，等待CPU调度
- setName 为线程设置名称
- getName 获取线程名称
- setDaemon 设置为后台线程或前台线程（默认）
- 如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止
- 如果是前台线程，主线程执行过程中，前台线程也进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止
- join 逐个执行每个线程，执行完毕后继续往下执行，该方法使的多线程变得无意义
- run 线程被CPU调度后执行Thread类对象的run方法
调用的方法：
直接调用
import threading
import time
def sayhi(num): # 定义每个线程要运行的函数
print("running on number:%s" %num)
time.sleep(3)
if __name__ == "__main__":
t_list = []
for i in range(10):
t = threading.Thread(target=sayhi,args=[i,])
t.start()
t_list.append(t)
for i in t_list:
i.join() # 阻断，等上面都执行完成后再执行后面的程序
print("---main---")
继承调用：
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()
线程锁（互斥锁Mutex）
一个进程下可以启动多个线程，多个线程共享父进程的内存空间，也就意味着每个线程可以访问同一份数据，这样就会出现2个线程同时要修改同一份数据。（3.x版本中不会出现这种情况）
未加锁的版本：
import time
import threading
def addNum():
global num #在每个线程中都获取这个全局变量
print(‘--get num:‘,num )
time.sleep(1)
num -=1 #对此公共变量进行-1操作
num = 100 #设定一个共享变量
thread_list = []
for i in range(100):
t = threading.Thread(target=addNum)
t.start()
thread_list.append(t)
for t in thread_list: #等待所有线程执行完毕
t.join()
print(‘final num:‘, num )
加锁版本：
import time
import threading
def addNum():
global num #在每个线程中都获取这个全局变量
print(‘--get num:‘,num )
time.sleep(1)
lock.acquire() #修改数据前加锁
num -=1 #对此公共变量进行-1操作
lock.release() #修改后释放
num = 100 #设定一个共享变量
thread_list = []
lock = threading.Lock() #生成全局锁
for i in range(100):
t = threading.Thread(target=addNum)
t.start()
thread_list.append(t)
for t in thread_list: #等待所有线程执行完毕
t.join()
print(‘final num:‘, num )
RLock（递归锁）
就是会有多层锁。
#!/usr/bin/env python3
import threading
def run1():
print("grab the first part data")
lock.acquire()
global num
num +=1
lock.release()
return num
def run2():
print("grab the second part data")
lock.acquire()
global num2
num2+=1
lock.release()
return num2
def run3():
lock.acquire()
res = run1()
print(‘--------between run1 and run2-----‘)
res2 = run2()
lock.release()
print(res,res2)
if __name__ == ‘__main__‘:
num,num2 = 0,0
lock = threading.RLock() #$ Rlock递归锁
for i in range(10):
t = threading.Thread(target=run3)
t.start()
while threading.active_count() != 1:
print(threading.active_count())
else:
print(‘----all threads done---‘)
print(num,num2)
Semaphore（信号量）
互斥锁同时只允许一个线程更改数据，而Semaphore是同时允许一定量的线程更改数据，比如去理发店，有3个理发师，最多允许3个人理发，后面的人只能排队等待。
import threading
import time
def run(n):
semaphore.acquire()
time.sleep(1)
print("run the thread: %s\n" %n)
semaphore.release()
if __name__ == ‘__main__‘:
num= 0
semaphore = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
for i in range(20):
t = threading.Thread(target=run,args=(i,))
t.start()
while threading.active_count() != 1:
pass #print threading.active_count()
else:
print(‘----all threads done---‘)
print(num)
Events
python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set , wait , clear
事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为False ，那么程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么 event.wait 方法便不阻塞
- clear：将“Flag”设置为False
- set：将“Flag”设置为True
通过event来实现两个或多个线程间的交互，下面是一个红绿灯的例子。
import threading
import time
import random
def light():
if not event.isSet():
event.set() #wait就不阻塞 #绿灯状态
count = 0
while True:
if count < 10:
print(‘\033[42;1m--green light on---\033[0m‘)
elif count <13:
print(‘\033[43;1m--yellow light on---\033[0m‘)
elif count <20:
if event.isSet():
event.clear()
print(‘\033[41;1m--red light on---\033[0m‘)
else:
count = 0
event.set() #打开绿灯
time.sleep(1)
count +=1
def car(n): #no bug version
while 1:
time.sleep(1)
if event.isSet(): #绿灯
print("car [%s] is running.." % n)
else:
print("car [%s] is waiting for the red light.." %n)
event.wait()
def car2(n):
while 1:
time.sleep(random.randrange(10))
if event.isSet(): #绿灯
print("car [%s] is running.." % n)
else:
print("car [%s] is waiting for the red light.." %n)
if __name__ == ‘__main__‘:
event = threading.Event()
Light = threading.Thread(target=light)
Light.start()
for i in range(3):
t = threading.Thread(target=car,args=(i,))
t.start()
生产着消费着模型
主要是基于单向队列queue，单向队列的遵循的是先进先出的原则
下面模拟一个吃包子的场景
import time,random
import queue,threading
q = queue.Queue()
def Producer(name):
count = 0
while count <20:
time.sleep(random.randrange(3))
q.put(count)
print(‘Producer %s has produced %s baozi..‘ %(name, count))
count +=1
def Consumer(name):
count = 0
while count <20:
time.sleep(random.randrange(4))
if not q.empty():
data = q.get()
print(data)
print(‘\033[32;1mConsumer %s has eat %s baozi...\033[0m‘ %(name, data))
else:
print("-----no baozi anymore----")
count +=1
p1 = threading.Thread(target=Producer, args=(‘A‘,))
c1 = threading.Thread(target=Consumer, args=(‘B‘,))
p1.start()
c1.start()
进程
多进程multiproceessing
from multiprocessing import Process
import os
def info(title):
print(title)
print(‘module name:‘, __name__)
print(‘parent process:‘, os.getppid()) #$ 父进程ID
print(‘process id:‘, os.getpid()) #$ 进程ID
print("\n\n")
def f(name):
info(‘\033[31;1mfunction f\033[0m‘)
print(‘hello‘, name)
if __name__ == ‘__main__‘:
info(‘\033[32;1mmain process line\033[0m‘)
p = Process(target=f, args=(‘bob‘,))
p.start()
p.join()
进程间通讯
不同进程间内存是不共享的，可以使用以下方法进行进程间的数据交换
Queues
使用方法和threading里的queue差不多
from multiprocessing import Process, Queue
def f(q):
q.put([42, None, ‘hello‘])
if __name__ == ‘__main__‘:
q = Queue()
p = Process(target=f, args=(q,))
p.start()
print(q.get()) # prints "[42, None, ‘hello‘]"
p.join()
Pipes
管道，双向的
from multiprocessing import Process, Pipe
def f(conn):
conn.send([42, None, ‘hello‘])
conn.close()
if __name__ == ‘__main__‘:
parent_conn, child_conn = Pipe()
p = Process(target=f, args=(child_conn,))
p.start()
print(parent_conn.recv()) # prints "[42, None, ‘hello‘]"
p.join()
Managers
from multiprocessing import Process, Manager
def f(d, l,n):
d[n] =n
d[‘2‘] = 2
d[0.25] = None
l.append(n)
print(l)
if __name__ == ‘__main__‘:
with Manager() as manager:
d = manager.dict()
l = manager.list(range(5))
p_list = []
for i in range(10):
p = Process(target=f, args=(d, l,i))
p.start()
p_list.append(p)
for res in p_list:
res.join()
print(d)
print(l)
进程同步
from multiprocessing import Process, Lock
def f(l, i):
l.acquire()
try:
print(‘hello world‘, i)
finally:
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
def Foo(i):
time.sleep(2)
return i+100
def Bar(arg):
print(‘-->exec done:‘,arg)
pool = Pool(5)
for i in range(10):
pool.apply_async(func=Foo, args=(i,),callback=Bar) #$ 异步
#pool.apply(func=Foo, args=(i,))
print(‘end‘)
pool.close()
pool.join()#进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。
时间： 2024-08-27 15:04:53