from threading import Thread money = 100 def task(): global money money -= 1 t_list = [] for i in range(100): t = Thread(target=task) t.start() t_list.append(t) # [线程1,线程2, 线程3 ...] for t in t_list: t.join() # 等待所有的线程运行结束 查看money是多少 print(money)
from threading import Thread import time money = 100 def task(): global money tmp = money time.sleep(0.1) money = tmp - 1 t_list = [] for i in range(100): t = Thread(target=task) t.start() t_list.append(t) # [线程1,线程2, 线程3 ...] for t in t_list: t.join() # 等待所有的线程运行结束 查看money是多少 print(money)
'''GIL不会影响程序层面的数据也不会保证它的修改是安全的要想保证得自己加锁''' from threading import Thread,Lock import time money = 100 mutex = Lock() def task(): mutex.acquire() global money tmp = money time.sleep(0.1) money = tmp - 1 mutex.release() t_list = [] for i in range(100): t = Thread(target=task) t.start() t_list.append(t) # [线程1 线程2 线程3 ... 线程100] for t in t_list: t.join() # 等待所有的线程运行结束 查看money是多少 print(money)
from threading import Thread from multiprocessing import Process import os import time def work(): # 计算密集型 res = 1 for i in range(1, 100000): res *= i if __name__ == '__main__': print(os.cpu_count()) # 12 查看当前计算机CPU个数 start_time = time.time() # p_list = [] # for i in range(12): # 一次性创建12个进程 # p = Process(target=work) # p.start() # p_list.append(p) # for p in p_list: # 确保所有的进程全部运行完毕 # p.join() t_list = [] for i in range(12): t = Thread(target=work) t.start() t_list.append(t) for t in t_list: t.join() print('总耗时:%s' % (time.time() - start_time)) # 获取总的耗时
2. IO密集型
def work(): time.sleep(2) # 模拟纯IO操作 if __name__ == '__main__': start_time = time.time() # t_list = [] # for i in range(100): # t = Thread(target=work) # t.start() # for t in t_list: # t.join() p_list = [] for i in range(100): p = Process(target=work) p.start() for p in p_list: p.join() print('总耗时:%s' % (time.time() - start_time))
from threading import Thread, Lock import time mutexA = Lock() # 类名加括号每执行一次就会产生一个新的对象 mutexB = Lock() # 类名加括号每执行一次就会产生一个新的对象 class MyThread(Thread): def run(self): self.func1() self.func2() def func1(self): mutexA.acquire() print(f'{self.name}抢到了A锁') mutexB.acquire() print(f'{self.name}抢到了B锁') mutexB.release() print(f'{self.name}释放了B锁') mutexA.release() print(f'{self.name}释放了A锁') def func2(self): mutexB.acquire() print(f'{self.name}抢到了B锁') time.sleep(1) mutexA.acquire() print(f'{self.name}抢到了A锁') mutexA.release() print(f'{self.name}释放了A锁') mutexB.release() print(f'{self.name}释放了B锁') for i in range(10): t = MyThread() t.start()
from threading import Thread, Lock, Semaphore import time import random sp = Semaphore(5) # 一次性产生五把锁 class MyThread(Thread): def run(self): sp.acquire() print(self.name) time.sleep(random.randint(1, 3)) sp.release() for i in range(20): t = MyThread() t.start()
from threading import Thread, Event import time event = Event() # 类似于造了一个红绿灯 def light(): print('红灯亮着的 所有人都不能动') time.sleep(3) print('绿灯亮了 油门踩到底 给我冲!!!') event.set() def car(name): print('%s正在等红灯' % name) event.wait() print('%s加油门 飙车了' % name) t = Thread(target=light) t.start() for i in range(5): t = Thread(target=car, args=('熊猫PRO%s' % i,)) t.start()
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from threading import current_thread import time pool = ThreadPoolExecutor(5) # 固定产生五个线程 def task(n): print(current_thread().name) print(n) time.sleep(1) for i in range(10): pool.submit(task,123) # 朝池子中提交任务(异步)
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from threading import current_thread import time pool = ThreadPoolExecutor(5) # 固定产生五个线程 def task(n): print(current_thread().name) # print(n) time.sleep(1) for i in range(10): res = pool.submit(task,123) # 朝池子中提交任务(异步) print(res.result()) # 同步操作
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from threading import current_thread import time pool = ThreadPoolExecutor(5) # 固定产生五个线程 def task(n): print(current_thread().name) # print(n) time.sleep(1) return '返回的结果' def func(*args, **kwargs): print('func', args, kwargs) print(args[0].result()) for i in range(10): # res = pool.submit(task,123) # 朝池子中提交任务(异步) # print(res.result()) # 同步操作 pool.submit(task, 123).add_done_callback(func) """异步回调:异步任务执行完成后有结果就会自动触发该机制"""
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from threading import current_thread import os import time # pool = ThreadPoolExecutor(5) # 固定产生五个线程 pool = ProcessPoolExecutor(5) # 固定产生五个进程 def task(n): # print(current_thread().name) print(os.getpid()) # 进程号 # print(n) time.sleep(1) return '返回的结果' def func(*args, **kwargs): print('func', args, kwargs) print(args[0].result()) if __name__ == '__main__': for i in range(20): # res = pool.submit(task,123) # 朝池子中提交任务(异步) # print(res.result()) # 同步操作 pool.submit(task, 123).add_done_callback(func) """异步回调:异步任务执行完成后有结果就会自动触发该机制"""
import time from gevent import monkey;monkey.patch_all() # 固定编写,用于检测所有的IO操作(猴子补丁) from gevent import spawn def func1(): print('func1 running') time.sleep(3) print('func1 over') def func2(): print('func2 running') time.sleep(5) print('func2 over') if __name__ == '__main__': start_time = time.time() # func1() # func2() s1 = spawn(func1) # 检测代码 一旦有IO自动切换(执行没有IO的操作,变相的等待IO结束) s2 = spawn(func2) s1.join() s2.join() print(time.time() - start_time) # 8.077142477035522 # 5.078031539916992
import socket from gevent import monkey;monkey.patch_all() from gevent import spawn def communication(sock): while True: data = sock.recv(1024) print(data.decode('utf8')) sock.send(data.upper()) def get_server(): server = socket.socket() server.bind(('127.0.0.1', 8080)) server.listen(5) while True: sock, addr = server.accept() # IO操作 spawn(communication, sock) s1 = spawn(get_server) s1.join()
import socket from threading import Thread,current_thread def get_client(): client = socket.socket() client.connect(('127.0.0.1', 8080)) while True: client.send(f'hello {current_thread().name}'.encode('utf8')) data = client.recv(1024) print(data.decode('utf8')) for i in range(400): t = Thread(target=get_client()) t.start()