Python Pool 多处理
宇宙之一粟 人气:0为什么要引入线程池
如果在程序中经常要用到线程,频繁的创建和销毁线程会浪费很多硬件资源,
所以需要把线程和任务分离。线程可以反复利用,省去了重复创建的麻烦。
在 Process 类中,我们必须显式地创建流程。但是,Pool 类更方便,您不必手动管理它。创建池对象的语法是 ?multiprocessing.Pool(processes, initializer, initargs, maxtasksperchild, context)?
? 。所有参数都是可选的。
- processes 表示您要创建的工作进程的数量。默认值通过 os.cpu_count() 获取。
- initializer第二个初始化器参数是一个用于初始化的函数。
- initargs 是传递给它的参数。
maxtasksperchild
表示分配给每个子进程的任务数。在完成该数量的任务之后,该进程将被一个新的工作进程替换。指定它的好处是任何未使用的资源都将被释放。如果未提供任何内容,则只要池存在,进程就会存在。
import time from multiprocessing import Pool def square(x): print(f"start process:{x}") square = x * x print(f"square {x}:{square}") time.sleep(1) print(f"end process:{x}") if __name__ == "__main__": starttime = time.time() pool = Pool() pool.map(square, range(0, 5)) pool.close() endtime = time.time() print(f"Time taken {endtime-starttime} seconds")
结果为:
start process:0
start process:1
square 1:1
square 0:0
end process:1
start process:2
end process:0
start process:3
square 2:4
square 3:9
end process:3
end process:2
start process:4
square 4:16
end process:4
Time taken 3.0474610328674316 seconds
在这里,我们从多处理模块中导入 Pool 类。在主函数中,我们创建了一个 Pool 类的对象。 pool.map() 将我们想要并行化的函数和一个可迭代的函数作为参数。它在可迭代的每个项目上运行给定的函数。它还接受一个可选的 chunksize 参数,它将可迭代对象拆分为等于给定大小的块,并将每个块作为单独的任务传递。 pool.close() 用于拒绝新任务。
我们可以看到花费的时间大约是 3 秒。
?pool.imap()?
? 与 ?pool.map()?
? 方法几乎相同。不同的是,每个项目的结果都是在准备好后立即收到的,而不是等待所有项目都完成。此外, ?map()?
? 方法将可迭代对象转换为列表(如果不是)。但是, ?imap()?
? 方法没有。
来看下一个例子:
import time from multiprocessing import Pool def square(x): print(f"start process {x}") square = x * x time.sleep(1) print(f"end process {x}") return square if __name__ == "__main__": pool = Pool() a = pool.map(square, range(0, 5)) print(a)
运行结果:
start process 0
start process 1
end process 0
start process 2
end process 1
start process 3
end process 2
start process 4
end process 3
end process 4
[0, 1, 4, 9, 16]
from concurrent.futures import ThreadPoolExecutor def say_hello(): print("Hello") executor = ThreadPoolExecutor(50) for i in range(0, 10): executor.submit(say_hello)
练习
利用 Python 多线程模拟商品秒杀过程,不可以出现超买和超卖的情况。假设A商品有50件参与秒杀活动,10分钟秒杀自动结束。
- kill_total 商品总数
- kill_num 成功抢购数
- kill_flag 有效标志位
- kill_user 成功抢购的用户ID
from redis_db import pool import redis import random from concurrent.futures import ThreadPoolExecutor s = set() while True: if len(s) == 1000: break num = random.randint(10000, 100000) s.add(num) print(s) con = redis.Redis( connection_pool=pool ) try: con.delete("kill_total", "kill_num", "kill_flag", "kill_user") con.set("kill_total", 50) con.set("kill_num", 0) con.set("kill_flag", 1) con.expire("kill_flag", 600) except Exception as e: print(e) finally: del con executor = ThreadPoolExecutor(200) def buy(): connection = redis.Redis( connection_pool=pool ) pipline = connection.pipline() try: if connection.exists("kill_flag") == 1: pipline.watch("kill_num", "kill_user") total = pipline.get("kill_total") num = int(pipline.get("kill_num").decode("utf-8")) if num < total: pipline.multi() pipline.incr("kill_num") user_id = s.pop() pipline.rpush("kill_user", user_id) pipline.execute() except Exception as e: print(e) finally: if "pipline" in dir(): pipline.reset() del connection for i in range(0, 1000): executor.submit(buy) print("秒杀活动已经结束")
加载全部内容