Thread 模块
import time
from threading import Thread
from multiprocessing import Process
#效率差别
def func(a):
a += 1
if __name__ == "__main__":
li = []
start = time.time()
for i in range(50):
p = Thread(target=func)
p.start()
li.append(p)
for i in li:
i.join()
pirnt(time.time()-start)
start = time.time()
li = []
for i in range(50):
t = Process(target=func)
t.start()
li.append(t)
for i in li:
i.join()
print(time.time()-start)
terminate(强制结束子进程) 在线程中没有
线程之间的数据共享
from threading import Thread
n = 100
def func():
global n
n -= 1
li = []
for i in range(100):
t = Thread(target=func)
t.start()
li.append(t)
for i in li:
i.join()
print(n)
守护线程
import time
from threading import Thread
def func():
while 1:
print(True)
time.sleep(0.5)
def inner():
print("in inner start")
time.sleep(3)
print("in inner end")
t1 = Thread(target=func)
t1.setDaemon(True)
t1.start()
t2 = Thread(target=inner)
t2.start()
time.sleep(1)
print("主进程")
主线程如果结束了,那么整个进程就结束了,守护线程会等待主线程结束之后才结束.
主进程 等待 守护进程 子进程 守护进程只守护主进程的代码就可以了
守护线程不行,主线程结束了那么整个进程就结束了,所有的线程就都结束了
例子 :
使用多线程实现tcp协议的socket server
客户端:
import socket
sk = socket.socket()
sk.connect_ex(("127.0.0.1",9090))
while 1:
sk.send(b"hello")
ret = sk.recv(1024).decode("utf-8")
print(ret)
sk.close()
服务器 :
import socket
from threading import Thread
sk = socket.socket()
sk.bind(("127.0.0.1",9090))
sk.listen()
def func(conn):
while 1:
ret = conn.recv(1024).decode("utf-8")
print(ret)
conn.send(b"world")
if __name__ == "__main__":
while 1:
conn,addr = sk.accept()
Thread(target=func,args=(conn,)).start()
from threading import Thread,get_ident
#开启线程的第二种方法和查看线程id
class Mythread(Thread):
def __init__(self,args):
super().__init__()
self.args = args
def run(self):
print("in my thread:",get_ident(),self.args)
print("main",get_ident())
t = Mythread("nana")
t.start()
线程中的方法
import time
from threading import Thread,get_ident,currentThread,enumerate,activeCount
class Mythread(Thread):
def __init__(self,args):
super().__init__()
self.args = args
def run(self):
time.sleep(0.1)
print(currentThread()) #返回当前的线程变量
print("in my thread:",get_ident(),self.args)
print("main:",get_ident())
t = Mythread("nana")
print(t.is_alive()) #返回线程是否是活动的
t.start()
t.setname("nana") #设置线程名称
print(t.getname()) #获取线程名称
print(activeCount()) #正在运行的线程的数量len(enumerate) #返回一个包含正在运行的线程list
print("t:",t)
Thread类 : 开启线程 传参数 join
和进程的差异 : 1,效率 2. 数据共享 3. 守护线程
面向对象的方式开启线程 :
thread对象的其他方法 : isAlive,setname,getname
threading模块的方法 : currentTread,activeCount,encumerate
GIL锁(全局解释器锁) : 锁线程, 同一个线程在同一时刻只能被一个CPU执行
在多进程/线程同时访问一个数据的时候就会产生数据不安全的现象.
多进程 访问文件
多线程 : 同时访问一个数据
GIL全局解释锁 : 在同一个进程里的每一个进程同一时间只能有一个线程访问CPU
尽量不要设置全局变量
只要在多线程/进程之间用到全局变量就加上锁
from threading import Thread,Lock
lock = Lock() #互斥锁
noodle = 100
def func(name):
global noodle
lock.acquire()
noodle -= 1
lock.release()
print("%s吃到面了"%name)
if __name__ == "__main__":
li = []
for i in range(100):
t = Thread(target=func)
t.start()
li.append(t)
for i in li:
i.join()
print(noodle)
科学家吃面问题(死锁现象)
import time
from threading import Thread,Lock
noodle = Lock()
fork = Lock()
def eat1(name):
noodle.acquire()
print("%s拿到面了"%name)
fork.acquire()
print("%s拿到叉子了"%name)
print("%s在吃面"%name)
time.sleep(0.5)
fork.release()
noodle.release()
def eat2(name):
fork.acquire()
print("%s拿到叉子了"%name)
noodle.acquire()
print("%s拿到面了"%name)
print("%s 在吃面"%name)
time.sleep(0.5)
noodle.release()
fork.release()
li = ["alex","egon","wusir","taibai"]
for i in li:
Thread(target=eat1,args=(i,)).start()
Thread(target=eat2,args=(i,)).start()
递归锁 :
from threading import RLock
lock = RLock()
lock.acquire()
print(55555)
lock.acquire()
print(666666)
lock.acquire()
print(99999)
递归锁解决死锁问题 :
import time
from threading import Thread,RLock
lock = RLock()
def eat1(name):
lock.acquire()
print("%s拿到面了"%name)
lock.acquire()
print("%s拿到了叉子了"%name)
print("%s 在吃饭"%name)
time.sleep(0.5)
lock.release()
lock.release()
def eat2(name):
lock.acquire()
print("%s拿到叉子了"%name)
lock.acquire()
print("%s拿到面了"%name)
time.sleep(0.5)
lock.release()
lock.release()
li = ["alex","egon","wusir","taibai"]
for i in li:
Thread(target=eat1,args=(i,)).start()
Thread(tarage=eat2,args=(i,)).start()
互斥锁解决死锁问题 :
import time
from threading import Thred,Lock
lock = Lock()
def eat1(name):
lock.acquire()
print("%s拿到面了"%name)
print("%s 拿到叉子了"%name)
print("%s在吃面"%name)
time.sleep(0.5)
lock.release()
def eat2(name):
lock.acquire()
print("%s拿到叉子了"%name)
print("%s拿到面了"%name)
print("%s 在吃面"%name)
time.sleep(0.5)
lock.release()
li = ["alex","egon","wusir","taibai"]
for i in li:
Thread(target=eat1,args=(i,)).start()
Thread(target=eat2,args=(i,)).start()
死锁 :
多把锁同时应用在多个线程中
互斥锁和递归锁哪个好?
递归锁 : 快速回复服务
死锁问题的出现,是程序的设计或者逻辑的问题
还应该进一步的排除和重构逻辑来保证使用互斥锁也不会发生死锁
互斥锁和递归锁的区别 :
互斥锁 : 就是在一个线程中不能连续多次acquire
递归锁 : 可以在同一个线程中acquire任意次,注意acquire多少次就需要release多少次
信号量 和 进程池
信号量就是 : 锁 + 计数器
import time
from multiprocessing import Semaphore,Process,Pool
def ktv(sem,i):
sem.acquire()
i += 1
sem.release()
def ktv1(i):
i += 1
if __name__ == "__main__":
start = time.time()
li = []
sem =Semaphore(5)
for i in range(100)
p = Process(targe=ktv,args=(sem,i))
p.start()
li.append(p)
for i in li:
i.join()
print(time.time()-start)
start = time.time()
p = Pool(5)
for i in range(100):
p.apply_async(func=ktv1,args=(i,))
p.close()
p.join()
print(time.time()-start)
进程池和信号量:
进程池 : 进程池里有几个进程就起几个,不管多少任务,池子里的进程个数是固定的,开启进程和关闭进程这些事都是需要固定的时间开销,就不产生额外的时间开销.且进程池中的进程数控制的好,那么操作系统的压力也小.
信号量 :
有多少个任务就起多少个进程/线程
可以帮助减少操作系统切换负担
但是并不能减少进/线程开启和关闭的时间
数据库连接 :
import time,random
from threading import Thread,Event
def check(e):
print("正在检测两台机器之间的网络情况...")
time.sleep(random.randint(1,3))
e.set()
def connect_db(e):
e.wait()
print("连接数据库..")
print("连接数据库成功....")
e = Event()
Thread(target=check,args=(e,)).start()
Thread(target=connect_db,args=(e,)).start()
import time,random
from threading import Thread,Event
def check(e):
print("正在检测两台机器之间的网络情况...")
time.sleep(random.randint(0,2))
e.set()
def connect_db(e):
n = 0
while n<3:
if e.is_set():
break
else:
e.wait(0.5)
n += 1
if n == 3:
raise TimeoutError
print("连接数据库....")
print("连接数据库成功...")
e = Event()
Thread(target=check,args=(e,)).start()
Thread(target=connect_db,args=(e,)).start()
条件 :
from threading import Condition
常用方法:
acquire , release , wait 阻塞 , notify (让wait解除阻塞的工具)
wait还是notify在执行这两个方法的前后,必须执行acquire和release
from threading import Condition ,Thread
def func(con,i):
con.acquire()
con.wait()
print("thread:",i)
con.release()
con = Condition()
for i in range(20):
Thread(target=func,args=(con,i)).start()
con.acquire()
con.notify_all() #帮助wait的子线程处理某个数据直到满足条件
con.release()
while 1:
num = int(input("<<<"))
con.acquire()
con.notify(num)
con.release()
定时器 :
from threading import Thread , Timer
def func():
print("我执行了")
Timer(1,func).start() #定时器
创建线程的时候,就规定他多久之后去执行