Python 并发编程之进程一.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "0c89b365",
   "metadata": {},
   "source": [
    "# Python 并发之进程一\n",
    "目录\n",
    "一、多进程的实现\n",
    "方法一\n",
    "方法二：\n",
    "二、使用进程的优缺点\n",
    "1、优点\n",
    "2、缺点\n",
    "三、进程的通信\n",
    "1、Queue 实现进程间通信\n",
    "2、Pipe 实现进程间通信（一边发送send(obj),一边接收(obj)）\n",
    "四、Manager管理器\n",
    "五、进程池\n",
    "总结\n",
    "\n",
    "一、多进程的实现\n",
    "\n",
    "方法一\n",
    "\n",
    "# 方法包装   多进程\n",
    "from multiprocessing import Process\n",
    "from time import sleep\n",
    "def func1(arg):\n",
    "    print(f'{arg}开始...')\n",
    "    sleep(2)\n",
    "    print(f'{arg}结束...')\n",
    "if __name__ == \"__main__\":\n",
    "    p1 = Process(target=func1,args=('p1',))\n",
    "    p2 = Process(target=func1,args=('p2',))\n",
    "    p1.start()\n",
    "    p2.start()\n",
    "\n",
    "方法二：\n",
    "\n",
    "二、使用进程的优缺点\n",
    "\n",
    "1、优点\n",
    "可以使用计算机多核，进行任务的并发执行，提高执行效率\n",
    "运行不受其他进程影响，创建方便\n",
    "空间独立，数据安全\n",
    "\n",
    "2、缺点\n",
    "进程的创建和删除消耗的系统资源较多\n",
    "\n",
    "三、进程的通信\n",
    "Python 提供了多种实现进程间通信的机制，主要有以下 2 种：\n",
    "\n",
    "1. Python multiprocessing 模块下的 Queue 类，提供了多个进程之间实现通信的诸多 方法\n",
    "\n",
    "2. Pipe，又被称为“管道”，常用于实现 2 个进程之间的通信，这 2 个进程分别位于管 道的两端\n",
    "\n",
    "\n",
    " \n",
    "Pipe 直译过来的意思是“管”或“管道”，该种实现多进程编程的方式，和实际生活中 的管（管道）是非常类似的。通常情况下，管道有 2 个口，而 Pipe 也常用来实现 2 个进程之 间的通信，这 2 个进程分别位于管道的两端，一端用来发送数据，另一端用来接收数据 - send(obj)\n",
    "\n",
    "发送一个 obj 给管道的另一端，另一端使用 recv() 方法接收。需要说明的是，该 obj 必 须是可序列化的，如果该对象序列化之后超过 32MB，则很可能会引发 ValueError 异常 - recv()\n",
    "\n",
    "接收另一端通过 send() 方法发送过来的数据 - close()\n",
    "\n",
    "关闭连接 - poll([timeout])\n",
    "\n",
    "返回连接中是否还有数据可以读取 - end_bytes(buffer[, offset[, size]])\n",
    "\n",
    "发送字节数据。如果没有指定 offset、size 参数，则默认发送 buffer 字节串的全部数 据；如果指定了 offset 和 size 参数，则只发送 buffer 字节串中从 offset 开始、长度为 size 的字节数据。通过该方法发送的数据，应该使用 recv_bytes() 或 recv_bytes_into 方法接收 - recv_bytes([maxlength])\n",
    "\n",
    "Python客栈送红包、纸质书\n",
    "\n",
    "接收通过 send_bytes() 方法发送的数据，maxlength 指定最多接收的字节数。该方法返 回接收到的字节数据 - recv_bytes_into(buffer[, offset])\n",
    "\n",
    "功能与 recv_bytes() 方法类似，只是该方法将接收到的数据放在 buffer 中\n",
    "\n",
    "\n",
    "1、Queue 实现进程间通信\n",
    "\n",
    "from multiprocessing import Process,current_process,Queue   # current_process 指的是当前进程\n",
    "# from queue import Queue\n",
    "import os\n",
    "def func(name,mq):\n",
    "    print('进程ID {} 获取了数据：{}'.format(os.getpid(),mq.get()))\n",
    "    mq.put('shiyi')\n",
    "if __name__ == \"__main__\":\n",
    "    # print('进程ID:{}'.format(current_process().pid))\n",
    "    # print('进程ID:{}'.format(os.getpid()))\n",
    "    mq = Queue()\n",
    "    mq.put('yangyang')\n",
    "    p1 = Process(target=func,args=('p1',mq))\n",
    "    p1.start()\n",
    "    p1.join()\n",
    "    print(mq.get())\n",
    "\n",
    "2、Pipe 实现进程间通信（一边发送send(obj),一边接收(obj)）\n",
    "\n",
    "from multiprocessing import Process,current_process,Pipe\n",
    "import os\n",
    "def func(name,con):\n",
    "    print('进程ID {} 获取了数据：{}'.format(os.getpid(),con.recv()))\n",
    "    con.send('你好！')\n",
    "if __name__ == \"__main__\":\n",
    "    # print('进程ID:{}'.format(current_process().pid))\n",
    "    con1,con2 = Pipe()\n",
    "    p1 = Process(target=func,args=('p1',con1))\n",
    "    p1.start()\n",
    "    con2.send(\"hello!\")\n",
    "    p1.join()\n",
    "    print(con2.recv())\n",
    "\n",
    "四、Manager管理器\n",
    "管理器提供了一种创建共享数据的方法，从而可以在不同进程中共享\n",
    "\n",
    "\n",
    "from multiprocessing import Process,current_process\n",
    "import os\n",
    "from multiprocessing import Manager\n",
    "def func(name,m_list,m_dict):\n",
    "    print('子进程ID {} 获取了数据：{}'.format(os.getpid(),m_list))\n",
    "    print('子进程ID {} 获取了数据：{}'.format(os.getpid(),m_dict))\n",
    "    m_list.append('你好')\n",
    "    m_dict['name'] = 'shiyi'   \n",
    "if __name__ == \"__main__\":\n",
    "    print('主进程ID:{}'.format(current_process().pid))\n",
    "    with Manager() as mgr:\n",
    "        m_list = mgr.list()\n",
    "        m_dict = mgr.dict()\n",
    "        m_list.append('Hello!!')\n",
    "        p1 = Process(target=func,args=('p1',m_list,m_dict))\n",
    "        p1.start()\n",
    "        p1.join()\n",
    "        print(m_list)\n",
    "        print(m_dict)\n",
    "\n",
    "五、进程池\n",
    "Python 提供了更好的管理多个进程的方式，就是使用进程池。\n",
    "\n",
    "进程池可以提供指定数量的进程给用户使用，即当有新的请求提交到进程池中时，如果池 未满，则会创建一个新的进程用来执行该请求；反之，如果池中的进程数已经达到规定最大 值，那么该请求就会等待，只要池中有进程空闲下来，该请求就能得到执行。\n",
    "\n",
    "使用进程池的优点\n",
    "\n",
    "1. 提高效率，节省开辟进程和开辟内存空间的时间及销毁进程的时间\n",
    "\n",
    "2. 节省内存空间\n",
    "\n",
    "类/方法\t功能\t参数\n",
    "Pool(processes)\n",
    "创建进程池对象\n",
    "processes 表示进程池\n",
    "中有多少进程\n",
    "pool.apply_async(func,a\n",
    "rgs,kwds)\n",
    "异步执行 ；将事件放入到进 程池队列\n",
    "func 事件函数\n",
    "args 以元组形式给\n",
    "func 传参\n",
    "kwds 以字典形式给\n",
    "func 传参 返回值：返\n",
    "回一个代表进程池事件的对\n",
    "象，通过返回值的 get 方法\n",
    "可以得到事件函数的返回值\n",
    "pool.apply(func,args,kw\n",
    "ds)\n",
    "同步执行；将事件放入到进程 池队列\n",
    "func 事件函数 args 以\n",
    "元组形式给 func 传参\n",
    "kwds 以字典形式给 func\n",
    "传参\n",
    "pool.close()\n",
    "关闭进程池\n",
    "pool.join()\n",
    "回收进程池\n",
    "pool.map(func,iter)\n",
    "类似于 python 的 map 函\n",
    "数，将要做的事件放入进程池\n",
    "func 要执行的函数\n",
    "iter 迭代对象\n",
    "\n",
    "from multiprocessing import Pool\n",
    "import os\n",
    "from time import sleep\n",
    "def func1(name):\n",
    "    print(f\"当前进程的ID:{os.getpid()},{name}\")\n",
    "    sleep(2)\n",
    "    return name\n",
    "def func2(args):\n",
    "    print(args)\n",
    "if __name__ == \"__main__\":\n",
    "    pool = Pool(5)\n",
    "    pool.apply_async(func = func1,args=('t1',),callback=func2)\n",
    "    pool.apply_async(func = func1,args=('t2',),callback=func2)\n",
    "    pool.apply_async(func = func1,args=('t3',),callback=func2)\n",
    "    pool.apply_async(func = func1,args=('t4',))\n",
    "    pool.apply_async(func = func1,args=('t5',))\n",
    "    pool.apply_async(func = func1,args=('t6',))\n",
    "    pool.close()\n",
    "    pool.join()\n",
    "\n",
    "from multiprocessing import Pool\n",
    "import os\n",
    "from time import sleep\n",
    "def func1(name):\n",
    "    print(f\"当前进程的ID:{os.getpid()},{name}\")\n",
    "    sleep(2)\n",
    "    return name\n",
    "if __name__ == \"__main__\":\n",
    "   with Pool(5) as pool:\n",
    "        args = pool.map(func1,('t1,','t2,','t3,','t4,','t5,','t6,','t7,','t8,'))\n",
    "        for a in args:\n",
    "            print(a)\n",
    "\n",
    "总结"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}