import os
# 注意一下，导入的是Process不是process（Class是大写开头）
from multiprocessing import Processdef test(name):print("[子进程-%s]PID：%d，PPID：%d" % (name, os.getpid(), os.getppid()))def main():print("[父进程]PID：%d，PPID：%d" % (os.getpid(), os.getppid()))p = Process(target=test, args=("萌萌哒", )) # 单个元素的元组表达别忘了(x,)p.start()p.join()  # 父进程回收子进程资源（内部调用了wait系列方法）if __name__ == '__main__':main()

import os
from multiprocessing import Processclass My_Process(Process):# 重写了Proce类的Init方法def __init__(self, name):self.__name = nameProcess.__init__(self)  # 调用父类方法# 重写了Process类的run()方法def run(self):print("[子进程-%s]PID：%d，PPID：%d" % (self.__name, os.getpid(),os.getppid()))def main():print("[父进程]PID：%d，PPID：%d" % (os.getpid(), os.getppid()))p = My_Process("萌萌哒") # 如果不指定方法，默认调Run方法p.start()p.join()  # 父进程回收子进程资源（内部调用了wait系列方法）if __name__ == '__main__':main()

# 3.4.x开始，Process有了一个BaseProcess
# https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/process.py
# https://github.com/lotapp/cpython3/tree/master/Lib/multiprocessing/process.py
def join(self, timeout=None):'''一直等到子进程over'''self._check_closed()# 断言（False就触发异常，提示就是后面的内容# 开发中用的比较多，部署的时候可以python3 -O xxx 去除所以断言assert self._parent_pid == os.getpid(), "只能 join 一个子进程"assert self._popen is not None, "只能加入一个已启动的进程"res = self._popen.wait(timeout) # 本质就是用了我们之前讲的wait系列if res is not None:_children.discard(self) # 销毁子进程

# 3.4.x开始，在popen_fork文件中（以前是multiprocessing.forking.py）
# https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/popen_fork.py
# https://github.com/lotapp/cpython3/tree/master/Lib/multiprocessing/popen_fork.py
def wait(self, timeout=None):if self.returncode is None:# 设置超时的一系列处理if timeout is not None:from multiprocessing.connection import waitif not wait([self.sentinel], timeout):return None# 核心操作return self.poll(os.WNOHANG if timeout == 0.0 else 0)return self.returncode# 回顾一下上次说的：os.WNOHANG - 如果没有子进程退出，则不阻塞waitpid()调用
def poll(self, flag=os.WNOHANG):if self.returncode is None:try:# 他的内部调用了waitpidpid, sts = os.waitpid(self.pid, flag)except OSError as e:# 子进程尚未创建# e.errno == errno.ECHILD == 10return Noneif pid == self.pid:if os.WIFSIGNALED(sts):self.returncode = -os.WTERMSIG(sts)else:assert os.WIFEXITED(sts), "Status is {:n}".format(sts)self.returncode = os.WEXITSTATUS(sts)return self.returncode

def test(a, b):assert b != 0, "哥哥，分母不能为0啊"return a / bdef main():test(1, 0)if __name__ == '__main__':main()

import os
import time
from multiprocessing import Pool  # 首字母大写def test(name):print("[子进程-%s]PID=%d，PPID=%d" % (name, os.getpid(), os.getppid()))time.sleep(1)def main():print("[父进程]PID=%d，PPID=%d" % (os.getpid(), os.getppid()))p = Pool(5) # 设置最多5个进程（不设置就默认为CPU核数）for i in range(10):# 异步执行p.apply_async(test, args=(i, )) # 同步用apply（如非必要不建议用）p.close() # 关闭池，不再加入新任务p.join() # 等待所有子进程执行完毕回收资源（join可以指定超时时间，eg：`p.join(1)`）print("over")if __name__ == '__main__':main()

# https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/pool.py
# https://github.com/lotapp/cpython3/tree/master/Lib/multiprocessing/pool.py
class Pool(object):def __init__(self, processes=指定的进程数,...):if processes is None:processes = os.cpu_count() or 1 # os.cpu_count() ~ CPU的核数

def apply_async(self,func,args=(),kwds={},callback=None,error_callback=None):if self._state != RUN:raise ValueError("Pool not running")result = ApplyResult(self._cache, callback, error_callback)self._taskqueue.put(([(result._job, 0, func, args, kwds)], None))return result

import os
import time
from multiprocessing import Pool  # 首字母大写def test(name):print("[子进程%s]PID=%d，PPID=%d" % (name, os.getpid(), os.getppid()))time.sleep(1)return namedef error_test(name):print("[子进程%s]PID=%d，PPID=%d" % (name, os.getpid(), os.getppid()))raise Exception("[子进程%s]啊，我挂了～" % name)def callback(result):"""成功之后的回调函数"""print("[子进程%s]执行完毕" % result)  # 没有返回值就为Nonedef error_callback(msg):"""错误之后的回调函数"""print(msg)def main():print("[父进程]PID=%d，PPID=%d" % (os.getpid(), os.getppid()))p = Pool()  # CPU默认核数for i in range(5):# 搞2个出错的看看if i > 2:p.apply_async(error_test,args=(i, ),callback=callback,error_callback=error_callback)  # 异步执行else:# 异步执行，成功后执行callback函数（有点像jq）p.apply_async(test, args=(i, ), callback=callback)p.close()  # 关闭池，不再加入新任务p.join()  # 等待所有子进程执行完毕回收资源print("over")if __name__ == '__main__':main()

import time
from multiprocessing import Pool, TimeoutErrordef test(x):"""开平方"""time.sleep(1)return x * xdef main():pool = Pool()task = pool.apply_async(test, (10, ))print(task)try:print(task.get(timeout=1))except TimeoutError as ex:print("超时了～", ex)if __name__ == '__main__':main()

import time
from multiprocessing import Pooldef test(x):return x * xif __name__ == '__main__':with Pool(processes=4) as pool:task = pool.apply_async(test, (10, ))print(task.get(timeout=1))obj_list = pool.map(test, range(10))print(obj_list)# 返回一个可迭代类的实例对象obj_iter = pool.imap(test, range(10))print(obj_iter)next(obj_iter)for i in obj_iter:print(i, end=" ")

class IMapIterator(object):def __init__(self, cache):self._cond = threading.Condition(threading.Lock())self._job = next(job_counter)self._cache = cacheself._items = collections.deque()self._index = 0self._length = Noneself._unsorted = {}cache[self._job] = selfdef __iter__(self):return self # 返回一个迭代器# 实现next方法def next(self, timeout=None):with self._cond:try:item = self._items.popleft()except IndexError:if self._index == self._length:raise StopIteration from Noneself._cond.wait(timeout)try:item = self._items.popleft()except IndexError:if self._index == self._length:raise StopIteration from Noneraise TimeoutError from Nonesuccess, value = itemif success:return valueraise value
......

import subprocessdef main():# os.execlp("ls", "ls", "-al")  # 执行Path环境变量可以搜索到的命令result = subprocess.run(["ls", "-al"])print(result)if __name__ == '__main__':main()

r"""
具有可访问I / O流的子进程
Subprocesses with accessible I/O streams此模块允许您生成进程，连接到它们输入/输出/错误管道，并获取其返回代码。
This module allows you to spawn processes, connect to their
input/output/error pipes, and obtain their return codes.完整文档可以查看：https://docs.python.org/3/library/subprocess.html
For a complete description of this module see the Python documentation.Main API
========
run(...): 运行命令，等待它完成，然后返回`CompletedProcess`实例。
Runs a command, waits for it to complete,
then returns a CompletedProcess instance.Popen(...): 用于在新进程中灵活执行命令的类
A class for flexibly executing a command in a new processConstants（常量）
---------
DEVNULL: 特殊值，表示应该使用`os.devnull`
Special value that indicates that os.devnull should be usedPIPE:    表示应创建`PIPE`管道的特殊值
Special value that indicates a pipe should be createdSTDOUT:  特殊值，表示`stderr`应该转到`stdout`
Special value that indicates that stderr should go to stdoutOlder API（尽量不用，说不定以后就淘汰了）
=========
call(...): 运行命令，等待它完成，然后返回返回码。
Runs a command, waits for it to complete, then returns the return code.check_call(...): Same as call() but raises CalledProcessError()
    if return code is not 0（返回值不是0就引发异常）check_output(...): 与check_call（）相同,但返回`stdout`的内容,而不是返回代码
Same as check_call but returns the contents of stdout instead of a return codegetoutput(...): 在shell中运行命令，等待它完成，然后返回输出
Runs a command in the shell, waits for it to complete,then returns the outputgetstatusoutput(...): 在shell中运行命令，等待它完成，然后返回一个（exitcode，output）元组
Runs a command in the shell, waits for it to complete,
then returns a (exitcode, output) tuple
"""

def run(*popenargs, input=None, capture_output=False,timeout=None, check=False, **kwargs):if input is not None:if 'stdin' in kwargs:raise ValueError('stdin和输入参数可能都不会被使用。')kwargs['stdin'] = PIPEif capture_output:if ('stdout' in kwargs) or ('stderr' in kwargs):raise ValueError('不能和capture_outpu一起使用stdout 或 stderr')kwargs['stdout'] = PIPEkwargs['stderr'] = PIPEwith Popen(*popenargs, **kwargs) as process:try:stdout, stderr = process.communicate(input, timeout=timeout)except TimeoutExpired:process.kill()stdout, stderr = process.communicate()raise TimeoutExpired(process.args, timeout, output=stdout, stderr=stderr)except:  # 包括KeyboardInterrupt的通信处理。process.kill()# 不用使用process.wait（），.__ exit__为我们做了这件事。raiseretcode = process.poll()if check and retcode:raise CalledProcessError(retcode, process.args, output=stdout, stderr=stderr)return CompletedProcess(process.args, retcode, stdout, stderr)

# https://github.com/lotapp/cpython3/blob/master/Lib/subprocess.py
class CompletedProcess(object):def __init__(self, args, returncode, stdout=None, stderr=None):self.args = argsself.returncode = returncodeself.stdout = stdoutself.stderr = stderrdef __repr__(self):"""对象按指定的格式显示"""args = ['args={!r}'.format(self.args),'returncode={!r}'.format(self.returncode)]if self.stdout is not None:args.append('stdout={!r}'.format(self.stdout))if self.stderr is not None:args.append('stderr={!r}'.format(self.stderr))return "{}({})".format(type(self).__name__, ', '.join(args))def check_returncode(self):"""如果退出代码非零，则引发CalledProcessError"""if self.returncode:raise CalledProcessError(self.returncode, self.args, self.stdout,self.stderr)

import subprocessdef main():result = subprocess.run(["ping", "www.baidu.com"])print(result.stdout)if __name__ == '__main__':main()

import subprocessdef main():process = subprocess.Popen(["ipython3"],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE)try:# 对pstree进行交互out, err = process.communicate(input=b'print("hello")', timeout=3)print("Out:%s\nErr:%s" % (out.decode(), err.decode()))except TimeoutError:# 如果超时到期，则子进程不会被终止，需要自己处理一下process.kill()out, err = process.communicate()print("Out:%s\nErr:%s" % (out.decode(), err.decode()))if __name__ == '__main__':main()

import subprocessdef main():# ps aux | grep bash# 进程1获取结果p1 = subprocess.Popen(["ps", "-aux"], stdout=subprocess.PIPE)# 得到进程1的结果再进行筛选p2 = subprocess.Popen(["grep", "bash"], stdin=p1.stdout, stdout=subprocess.PIPE)# 关闭写段（结果已经获取到进程2中了，防止干扰显示）p1.stdout.close()# 与流程交互：将数据发送到stdin并关闭它。msg_tuple = p2.communicate()# 输出结果print(msg_tuple[0].decode())if __name__ == '__main__':main()

from multiprocessing import Process, Pipedef test(w):w.send("[子进程]老爸，老妈回来记得喊我一下～")msg = w.recv()print(msg)def main():r, w = Pipe()p1 = Process(target=test, args=(w, ))p1.start()msg = r.recv()print(msg)r.send("[父进程]滚犊子，赶紧写作业，不然我得跪方便面！")p1.join()if __name__ == '__main__':main()

# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/context.py
def Pipe(self, duplex=True):'''返回由管道连接的两个连接对象'''from .connection import Pipereturn Pipe(duplex)

# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/connection.py
if sys.platform != 'win32':def Pipe(duplex=True):'''返回管道两端的一对连接对象'''if duplex:# 双工内部其实是socket系列（下次讲）s1, s2 = socket.socketpair()s1.setblocking(True)s2.setblocking(True)c1 = Connection(s1.detach())c2 = Connection(s2.detach())else:# 这部分就是我们上次讲的pipe管道fd1, fd2 = os.pipe()c1 = Connection(fd1, writable=False)c2 = Connection(fd2, readable=False)return c1, c2
else: def Pipe(duplex=True):# win平台的一系列处理......c1 = PipeConnection(h1, writable=duplex)c2 = PipeConnection(h2, readable=duplex)return c1, c2

from multiprocessing import Process, Pipedef test(w):# 只能写w.send("[子进程]老爸，咱们完了，老妈一直在门口～")def main():r, w = Pipe(duplex=False)p1 = Process(target=test, args=(w, ))p1.start() # 你把这个放在join前面就直接死锁了msg = r.recv() # 只能读print(msg)p1.join()if __name__ == '__main__':main()

from multiprocessing import Pipe, Pooldef proc_test1(conn):conn.send("[小明]小张，今天哥们要见一女孩，你陪我呗，我24h等你回复哦～")msg = conn.recv()print(msg)def proc_test2(conn):msg = conn.recv()print(msg)conn.send("[小张]不去，万一被我帅气的外表迷倒就坑了～")def main():conn1, conn2 = Pipe()p = Pool()p.apply_async(proc_test1, (conn1, ))p.apply_async(proc_test2, (conn2, ))p.close()  # 关闭池，不再接收新任务p.join()  # 等待回收，必须先关才能join，不然会异常if __name__ == '__main__':main()

[小明]小张，今天哥们要见一女孩，你陪我呗，我24h等你回复哦～
[小张]不去，万一被我帅气的外表迷倒就坑了～

# https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/pool.py
# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/pool.py
def join(self):util.debug('joining pool')if self._state == RUN:# 没关闭就join，这边就会抛出一个异常raise ValueError("Pool is still running")elif self._state not in (CLOSE, TERMINATE):raise ValueError("In unknown state")self._worker_handler.join()self._task_handler.join()self._result_handler.join()for p in self._pool:p.join() # 循环join回收

self._processes = processes # 指定的进程数
self._pool = [] # 列表
self._repopulate_pool() # 给列表append内容的方法

def _repopulate_pool(self):# 指定进程数-当前进程数，差几个补几个for i in range(self._processes - len(self._pool)):w = self.Process(target=worker,args=(self._inqueue, self._outqueue,self._initializer,self._initargs, self._maxtasksperchild,self._wrap_exception))self._pool.append(w) # 重点来了w.name = w.name.replace('Process', 'PoolWorker')w.daemon = True # pool退出后，通过pool创建的进程都会退出w.start()util.debug('added worker')

from multiprocessing import Process, Queuedef test(q):q.put("[子进程]老爸，我出去嗨了")print(q.get())def main():q = Queue()p = Process(target=test, args=(q, ))p.start()msg = q.get()print(msg)q.put("[父进程]去吧比卡丘～")p.join()if __name__ == '__main__':main()

# 队列类型，使用PIPE，缓存，线程
class Queue(object):# ctx = multiprocessing.get_context("xxx")# 上下文总共3种：spawn、fork、forkserver（扩展部分会提一下）def __init__(self, maxsize=0, *, ctx):# 默认使用最大容量if maxsize <= 0:from .synchronize import SEM_VALUE_MAX as maxsizeself._maxsize = maxsize  # 指定队列大小# 创建了一个PIPE匿名管道（单向）self._reader, self._writer = connection.Pipe(duplex=False)# `multiprocessing/synchronize.py > Lock`self._rlock = ctx.Lock()  # 进程锁（读）【非递归】self._opid = os.getpid()  # 获取PIDif sys.platform == 'win32':self._wlock = Noneelse:self._wlock = ctx.Lock()  # 进程锁（写）【非递归】# Semaphore信号量通常用于保护容量有限的资源# 控制信号量,超了就异常self._sem = ctx.BoundedSemaphore(maxsize)# 不忽略PIPE管道破裂的错误self._ignore_epipe = False # 线程相关操作self._after_fork()# 向`_afterfork_registry`字典中注册if sys.platform != 'win32':register_after_fork(self, Queue._after_fork)

def get(self, block=True, timeout=None):# 默认情况是阻塞（lock加锁）if block and timeout is None:with self._rlock:res = self._recv_bytes()self._sem.release()  # 信号量+1else:if block:deadline = time.monotonic() + timeout# 超时抛异常if not self._rlock.acquire(block, timeout):raise Emptytry:if block:timeout = deadline - time.monotonic()# 不管有没有内容都去读，超时就抛异常if not self._poll(timeout):raise Emptyelif not self._poll():raise Empty# 接收字节数据作为字节对象res = self._recv_bytes()self._sem.release()  # 信号量+1finally:# 释放锁self._rlock.release()# 释放锁后，重新序列化数据return _ForkingPickler.loads(res)

def put(self, obj, block=True, timeout=None):# 如果Queue已经关闭就抛异常assert not self._closed, "Queue {0!r} has been closed".format(self)# 记录信号量的锁if not self._sem.acquire(block, timeout):raise Full  # 超过数量，抛个异常# 条件变量允许一个或多个线程等待，直到另一个线程通知它们with self._notempty:if self._thread is None:self._start_thread()self._buffer.append(obj)self._notempty.notify()

def get_nowait(self):return self.get(False)def put_nowait(self, obj):return self.put(obj, False)

from multiprocessing import Queuedef main():q = Queue(3)  # 只能 put 3条消息q.put([1, 2, 3, 4])  # put一个List类型的消息q.put({"a": 1, "b": 2})  # put一个Dict类型的消息q.put({1, 2, 3, 4})  # put一个Set类型的消息try:# 不加timeout，就一直阻塞，等消息队列有空位才能发出去q.put("再加条消息呗", timeout=2)# Full(Exception)是空实现，你可以直接用Exceptionexcept Exception:print("消息队列已满，队列数%s，当前存在%s条消息" % (q._maxsize, q.qsize()))try:# 非阻塞，不能put就抛异常q.put_nowait("再加条消息呗")  # 相当于q.put(obj,False)except Exception:print("消息队列已满，队列数%s，当前存在%s条消息" % (q._maxsize, q.qsize()))while not q.empty():print("队列数：%s，当前存在%s条消息 内容%s" % (q._maxsize, q.qsize(), q.get_nowait()))print("队列数：%s，当前存在：%s条消息" % (q._maxsize, q.qsize()))if __name__ == '__main__':main()

import os
import time
from multiprocessing import Process, Queuedef pro_test1(q):print("[子进程1]PPID=%d,PID=%d,GID=%d"%(os.getppid(), os.getpid(), os.getgid()))q.put("[子进程1]小明，今晚撸串不？")# 设置一个简版的重试机制（三次重试）for i in range(3):if not q.empty():print(q.get())breakelse:time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6sdef pro_test2(q):print("[子进程2]PPID=%d,PID=%d,GID=%d"%(os.getppid(), os.getpid(), os.getgid()))print(q.get())time.sleep(4)  # 模拟一下网络延迟q.put("[子进程2]不去，我今天约了妹子")def main():queue = Queue()p1 = Process(target=pro_test1, args=(queue, ))p2 = Process(target=pro_test2, args=(queue, ))p1.start()p2.start()p1.join()p2.join()if __name__ == '__main__':main()

import os
import time
from multiprocessing import Pool, Queuedef error_callback(msg):print(msg)def pro_test1(q):print("[子进程1]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))q.put("[子进程1]小明，今晚撸串不？")# 设置一个简版的重试机制（三次重试）for i in range(3):if not q.empty():print(q.get())breakelse:time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6sdef pro_test2(q):print("[子进程2]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))print(q.get())time.sleep(4)  # 模拟一下网络延迟q.put("[子进程2]不去，我今天约了妹子")def main():print("[父进程]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))queue = Queue()p = Pool()p.apply_async(pro_test1, args=(queue, ), error_callback=error_callback)p.apply_async(pro_test2, args=(queue, ), error_callback=error_callback)p.close()p.join()if __name__ == '__main__':main()

import os
import time
from multiprocessing import Pool, Managerdef error_callback(msg):print(msg)def pro_test1(q):print("[子进程1]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))q.put("[子进程1]小明，今晚撸串不？")# 设置一个简版的重试机制（三次重试）for i in range(3):if not q.empty():print(q.get())breakelse:time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6sdef pro_test2(q):print("[子进程2]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))print(q.get())time.sleep(4)  # 模拟一下网络延迟q.put("[子进程2]不去，我今天约了妹子")def main():print("[父进程]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))queue = Manager().Queue()p = Pool()p.apply_async(pro_test1, args=(queue, ), error_callback=error_callback)p.apply_async(pro_test2, args=(queue, ), error_callback=error_callback)p.close()p.join()if __name__ == '__main__':main()

import os
import time
from multiprocessing import Pool, Queuedef error_callback(msg):print(msg)q = Queue()def pro_test1():global qprint("[子进程1]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))q.put("[子进程1]小明，今晚撸串不？")# 设置一个简版的重试机制（三次重试）for i in range(3):if not q.empty():print(q.get())breakelse:time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6sdef pro_test2():global qprint("[子进程2]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))print(q.get())time.sleep(4)  # 模拟一下网络延迟q.put("[子进程2]不去，我今天约了妹子")def main():print("[父进程]PPID=%d,PID=%d,GID=%d" % (os.getppid(), os.getpid(),os.getgid()))q = Queue()p = Pool()p.apply_async(pro_test1, error_callback=error_callback)p.apply_async(pro_test2, error_callback=error_callback)p.close()p.join()if __name__ == '__main__':main()

import multiprocessing as mpdef foo(q):q.put('hello')if __name__ == '__main__':mp.set_start_method('spawn') # 不要过多使用q = mp.Queue()p = mp.Process(target=foo, args=(q,))p.start()print(q.get())p.join()

import multiprocessing as mpdef foo(q):q.put('hello')if __name__ == '__main__':ctx = mp.get_context('spawn')q = ctx.Queue()p = ctx.Process(target=foo, args=(q,))p.start()print(q.get())p.join()

# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/context.py
def log_to_stderr(self, level=None):'''打开日志记录并添加一个打印到stderr的处理程序'''from .util import log_to_stderrreturn log_to_stderr(level)

# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/util.py
def log_to_stderr(level=None):'''打开日志记录并添加一个打印到stderr的处理程序'''# 全局变量默认是Falseglobal _log_to_stderrimport logging# 日记记录转换成文本formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)# 一个处理程序类，它将已适当格式化的日志记录写入流handler = logging.StreamHandler()  # 此类不会关闭流，因为用到了sys.stdout|sys.stderr# 设置格式：'[%(levelname)s/%(processName)s] %(message)s'handler.setFormatter(formatter)# 返回`multiprocessing`专用的记录器logger = get_logger()# 添加处理程序logger.addHandler(handler)if level:# 设置日记级别logger.setLevel(level)# 现在log是输出到stderr的_log_to_stderr = Truereturn _logger

import logging
from multiprocessing import Process, log_to_stderrdef test():print("test")def start_log():# 把日记输出定向到sys.stderr中logger = log_to_stderr()# 设置日记记录级别# 敏感程度：DEBUG、INFO、WARN、ERROR、CRITICALprint(logging.WARN == logging.WARNING)  # 这两个是一样的level = logging.INFOlogger.setLevel(level)  # 设置日记级别(一般都是WARN)# 自定义输出# def log(self, level, msg, *args, **kwargs):logger.log(level, "我是通用格式")  # 通用，下面的内部也是调用的这个logger.info("info 测试")logger.warning("warning 测试")logger.error("error 测试")def main():start_log()# 做的操作都会被记录下来p = Process(target=test)p.start()p.join()if __name__ == '__main__':main()

# https://github.com/lotapp/cpython3/blob/master/Lib/multiprocessing/managers.py
class Value(object):def __init__(self, typecode, value, lock=True):self._typecode = typecodeself._value = valuedef get(self):return self._valuedef set(self, value):self._value = valuedef __repr__(self):return '%s(%r, %r)' % (type(self).__name__, self._typecode, self._value)value = property(get, set) # 给value设置get和set方法（和value的属性装饰器一样效果）def Array(typecode, sequence, lock=True):return array.array(typecode, sequence)

from multiprocessing import Process, Value, Arraydef proc_test1(value, array):print("子进程1", value.value)array[0] = 10print("子进程1", array[:])def proc_test2(value, array):print("子进程2", value.value)array[1] = 10print("子进程2", array[:])def main():try:value = Value("d", 3.14)  # d 类型，相当于C里面的doublearray = Array("i", range(10))  # i 类型，相当于C里面的intprint(type(value))print(type(array))p1 = Process(target=proc_test1, args=(value, array))p2 = Process(target=proc_test2, args=(value, array))p1.start()p2.start()p1.join()p2.join()print("父进程", value.value)  # 获取值print("父进程", array[:])  # 获取值except Exception as ex:print(ex)else:print("No Except")if __name__ == '__main__':main()

<class 'multiprocessing.sharedctypes.Synchronized'>
<class 'multiprocessing.sharedctypes.SynchronizedArray'>
子进程1 3.14
子进程1 [10, 1, 2, 3, 4, 5, 6, 7, 8, 9]
子进程2 3.14
子进程2 [10, 10, 2, 3, 4, 5, 6, 7, 8, 9]
父进程 3.14
父进程 [10, 10, 2, 3, 4, 5, 6, 7, 8, 9]
No Except

typecode_to_type = {'c': ctypes.c_char,'u': ctypes.c_wchar,'b': ctypes.c_byte,'B': ctypes.c_ubyte,'h': ctypes.c_short,'H': ctypes.c_ushort,'i': ctypes.c_int,'I': ctypes.c_uint,'l': ctypes.c_long,'L': ctypes.c_ulong,'q': ctypes.c_longlong,'Q': ctypes.c_ulonglong,'f': ctypes.c_float,'d': ctypes.c_double
}

from multiprocessing import Process, Valuedef proc_test1(value):for i in range(1000):value.value += 1def main():value = Value("i", 0)p_list = [Process(target=proc_test1, args=(value, )) for i in range(5)]# 批量启动for i in p_list:i.start()# 批量资源回收for i in p_list:i.join()print(value.value)if __name__ == '__main__':main()

# 通用方法
def proc_test1(value):for i in range(1000):if value.acquire():value.value += 1value.release()# 官方案例：(Lock可以使用with托管)
def proc_test1(value):for i in range(1000):with value.get_lock():value.value += 1# 更多可以查看：`sharedctypes.SynchronizedBase` 源码

def Value(typecode_or_type, *args, lock=True, ctx=None):'''返回Value的同步包装器'''obj = RawValue(typecode_or_type, *args)if lock is False:return obj# 默认支持Lockif lock in (True, None):ctx = ctx or get_context() # 获取上下文lock = ctx.RLock() # 获取递归锁if not hasattr(lock, 'acquire'): raise AttributeError("%r has no method 'acquire'" % lock)# 一系列处理return synchronized(obj, lock, ctx=ctx)def Array(typecode_or_type, size_or_initializer, *, lock=True, ctx=None):'''返回RawArray的同步包装器'''obj = RawArray(typecode_or_type, size_or_initializer)if lock is False:return obj# 默认是支持Lock的if lock in (True, None):ctx = ctx or get_context() # 获取上下文lock = ctx.RLock()  # 递归锁属性# 查看是否有acquire属性if not hasattr(lock, 'acquire'):raise AttributeError("%r has no method 'acquire'" % lock)return synchronized(obj, lock, ctx=ctx)

from multiprocessing import Pool, Managerdef test1(d, l):d[1] = '1'd['2'] = 2d[0.25] = Nonel.reverse()def test2(d, l):print(d)print(l)def main():with Manager() as manager:dict_test = manager.dict()list_test = manager.list(range(10))pool = Pool()pool.apply_async(test1, args=(dict_test, list_test))pool.apply_async(test2, args=(dict_test, list_test))pool.close()pool.join()if __name__ == '__main__':main()

from multiprocessing import Manager, Process, Lockdef test(dict1, lock):for i in range(100):with lock:  # 你可以把这句话注释掉，然后就知道为什么加了dict1["year"] += 1def main():with Manager() as m:lock = Lock()dict1 = m.dict({"year": 2000})p_list = [Process(target=test, args=(dict1, lock)) for i in range(5)]for i in p_list:i.start()for i in p_list:i.join()print(dict1)if __name__ == '__main__':main()

# processes：进程数
# initializer,initargs 初始化进行的操作
# maxtaskperchild：每个进程执行task的最大数目
# contex：上下文对象
def __init__(self, processes=None, initializer=None, initargs=(),maxtasksperchild=None, context=None):

import os
import time
from multiprocessing import Pool, Queuedef error_callback(msg):print(msg)def pro_test1():print("[子进程1]PPID=%d,PID=%d" % (os.getppid(), os.getpid()))q.put("[子进程1]小明，今晚撸串不？")# 设置一个简版的重试机制（三次重试）for i in range(3):if not q.empty():print(q.get())breakelse:time.sleep((i + 1) * 2)  # 第一次1s，第二次4s，第三次6sdef pro_test2():print("[子进程2]PPID=%d,PID=%d" % (os.getppid(), os.getpid()))print(q.get())time.sleep(4)  # 模拟一下网络延迟q.put("[子进程2]不去，我今天约了妹子")def init(queue):global qq = queuedef main():print("[父进程]PPID=%d,PID=%d" % (os.getppid(), os.getpid()))queue = Queue()p = Pool(initializer=init, initargs=(queue, ))p.apply_async(pro_test1, error_callback=error_callback)p.apply_async(pro_test2, error_callback=error_callback)p.close()p.join()if __name__ == '__main__':main()

from multiprocessing import Queue
from multiprocessing.managers import BaseManagerdef main():# 用来身份验证的key = b"8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92"get_zhang_queue = Queue()  # 小张消息队列get_ming_queue = Queue()  # 小明消息队列# 把Queue注册到网络上, callable参数关联了Queue对象BaseManager.register("get_zhang_queue", callable=lambda: get_zhang_queue)BaseManager.register("get_ming_queue", callable=lambda: get_ming_queue)# 实例化一个Manager对象。绑定ip+端口, 设置验证秘钥manager = BaseManager(address=("192.168.36.235", 5438), authkey=key)# 运行servemanager.get_server().serve_forever()if __name__ == '__main__':main()

from multiprocessing.managers import BaseManagerdef main():"""客户端1"""key = b"8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92"# 注册对应方法的名字（从网络上获取Queue）BaseManager.register("get_ming_queue")BaseManager.register("get_zhang_queue")# 实例化一个Manager对象。绑定ip+端口, 设置验证秘钥m = BaseManager(address=("192.168.36.235", 5438), authkey=key)# 连接到服务器m.connect()q1 = m.get_zhang_queue()  # 在自己队列里面留言q1.put("[小张]小明，老大明天是不是去外地办事啊？")q2 = m.get_ming_queue()  # 获取小明说的话print(q2.get())if __name__ == '__main__':main()

/// <summary>
/// 模拟一个网络操作（别忘了重试机制）
/// </summary>
/// <param name="url">url</param>
/// <returns></returns>
private async static Task<string> DownloadStringAsync(string url)
{using (var client = new HttpClient()){// 设置第一次重试时间var nextDelay = TimeSpan.FromSeconds(1);for (int i = 0; i < 3; i++){try{return await client.GetStringAsync(url);}catch { }await Task.Delay(nextDelay); // 用异步阻塞的方式防止服务器被太多重试给阻塞了nextDelay *= 2; // 3次重试机会，第一次1s，第二次2s，第三次4s}// 最后一次尝试，错误就抛出return await client.GetStringAsync(url);}
}

/// <summary>
/// 1.批量任务
/// </summary>
/// <param name="list"></param>
/// <returns></returns>
private async static Task<string[]> DownloadStringAsync(IEnumerable<string> list)
{using (var client = new HttpClient()){var tasks = list.Select(url => client.GetStringAsync(url)).ToArray();return await Task.WhenAll(tasks);}
}

/// <summary>
/// 2.返回首先完成的Task
/// </summary>
/// <param name="list"></param>
/// <returns></returns>
private static async Task<string> GetIPAsync(IEnumerable<string> list)
{using (var client = new HttpClient()){var tasks = list.Select(url => client.GetStringAsync(url)).ToArray();var task = await Task.WhenAny(tasks); // 返回第一个完成的Taskreturn await task;}
}

/// <summary>
/// 3.超时取消
/// </summary>
/// <returns></returns>
private static async Task<string> CancellMethod()
{//实例化取消任务var cts = new CancellationTokenSource();cts.CancelAfter(TimeSpan.FromSeconds(3)); // 设置失效时间为3stry{return await DoSomethingAsync(cts.Token);}// 任务已经取消会引发TaskCanceledExceptioncatch (TaskCanceledException ex){return "false";}
}
/// <summary>
/// 模仿一个耗时操作
/// </summary>
/// <returns></returns>
private static async Task<string> DoSomethingAsync(CancellationToken token)
{await Task.Delay(TimeSpan.FromSeconds(5), token);return "ok";
}

static void Main(string[] args)
{IEnumerable<int> list = new List<int>() { 1, 2, 3, 4, 5, 7, 8, 9 };foreach (var item in ParallelMethod(list)){Console.WriteLine(item);}
}
/// <summary>
/// 举个例子
/// </summary>
private static IEnumerable<int> ParallelMethod(IEnumerable<int> list)
{return list.AsParallel().Select(x => x * x);
}

/// <summary>
/// 举个例子
/// </summary>
private static IEnumerable<int> ParallelMethod(IEnumerable<int> list)
{return list.AsParallel().AsOrdered().Select(x => x * x);
}

var list = new List<long>();
for (long i = 0; i < 1000000; i++)
{list.Add(i);
}
Console.WriteLine(GetSumParallel(list));

private static long GetSumParallel(IEnumerable<long> list)
{return list.AsParallel().Sum();
}

await Task.Run(() =>Parallel.Invoke(() => Task.Delay(TimeSpan.FromSeconds(3)),() => Task.Delay(TimeSpan.FromSeconds(2))));

Parallel.Invoke(new ParallelOptions() { CancellationToken = token }, actions);