[Python知识库]快速的Python：一文打通thread和process

让你的python快起来

Python作为一种解释型语言，没有预编译的过程，相对于其他的编程语言来说，是比较慢的。
为了让你的python代码变得更快速，你可以采用线程，进程，协程的方式来加快代码的运行速度.

1.线程

线程:是计算机CPU进行调度的最小单元

先指出一个问题:因为cpython在开发之初引入了一个叫GIL的全局解释器锁，导致python的多线程，实际上每个进程都只有一个线程被cpu调度
有点像单线程的模式.那大家就会疑惑 引入了GIL锁 ，那python得多线程还有什么用呢?

下面用白话给不太了解的朋友介绍一下:
例如你使用爬虫去爬取数据的过程中会有大部分的时间用来网络io,
等待服务器返回响应的过程，你是可以做其他事的,GIL会在这个线程执行一会就让另一个线程在执行一会
在等待的时候是做了事情的

所以Python的线程是适合做这些有io等待的操作.
下面介绍Python中的线程使用常见场景:
1.基本使用

import threading
from threading import Thread
import time
def func():
    print(threading.current_thread().name)
    time.sleep(2)


for i in range(10):
    thread_each=Thread(target=func,name="thread_%d"%i)
    thread_each.start()

2,自定义线程类

from threading import Thread
import time
def fun2(name):
    print("say thread11")
    print(name,'children')
class Mythread(Thread):
    def run(self):
                """
        在这里确定你的任务要执行什么,
        self._target可以不传
        直接贴出源码
                try:
            if self._target:
                self._target(*self._args, **self._kwargs)
        finally:
            # Avoid a refcycle if the thread is running a function with
            # an argument that has a member that points to the thread.
            del self._target, self._args, self._kwargs
        :return:
        """
        print("i am time over!!")
        self._target(*self._args,**self._kwargs)

for i in range(10):
    thread_each=Mythread(target=func,name="thread_%d"%i)
    thread_each.start()

3.线程池
创建一个池子，限制池子里面有几个线程，池子中没有剩余线程的时候处于等待,等待池子里面有空闲的进程

from concurrent.futures import ThreadPoolExecutor
import time
def func():
    print(threading.current_thread().name)
    time.sleep(2)
with ThreadPoolExecutor(max_workers=4) as pool:
    for i in range(10):
        pool.submit(func)
    pool.shutdown(True)  #等待进程池中的任务全部执行

print("main thread over!!!")

给线程池里面的任务加上一个回调

import time
from concurrent.futures import ThreadPoolExecutor
def task(sle):
    print(sle)
    time.sleep(sle)
    return {
        "name":"xxx",
        "pid":9978
    }

def jsonx(res):
    print(res.result())
with ThreadPoolExecutor(max_workers=4) as pool:
    for i in range(4):
        future=pool.submit(task,3)
        future.add_done_callback(jsonx)

假抢票的案例
注意：global的使用

import threading
from concurrent.futures import ThreadPoolExecutor
file=open('./num.txt','r+',encoding='utf-8')
lock=threading.RLock()
num = int(file.read())
def qiangPiao():
    lock.acquire()
    global num
    num=num-1
    print("inner",num)
    file.write(str(num)+'\n')
    lock.release()

with ThreadPoolExecutor(max_workers=4) as pool:
     for i in range(10):
        pool.submit(qiangPiao)
     pool.shutdown(True)

#shutdown true会释放掉所有的资源

#单例设计模式
import time
from threading import  Thread
class Text(object):
    instance=None
    def __init__(self):
        pass

    def __new__(cls, *args, **kwargs):
        if not  cls.instance:
            print("enter1")
            time.sleep(2)
            instance=object.__new__(cls, *args, **kwargs)
            cls.instance=instance
            return cls.instance
        else:
            return  cls.instance


if __name__ == '__main__':
    for i in range(10):
        task=Thread(target=Text)
        task.start()

2.进程

进程是计算机进行资源分配的最小单位

#查看你的cpu有几个
print(multiprocessing.cpu_count())

python是一门可以移植的语言
根据platform的不同,三大执行模式

fork (linux 内核的) 会拷贝当前任务的所有所需资源 支持传递 线程锁和文件对象

spawn (windows 和mac) 不支持传递文件对象和线程锁 注mac系统本来是支持fork函数,但是python解释器设置的spawn模式
(spawn 父进程开启一个Python解释器子进程，只会拷贝一些必要的资源)

forkserver 不支持传递文件对象和线程锁
(创建一个process server ，每次创建新进程，就像他请求)
后面两种模式都必须创建相应的重要参数

下面举三个例子看看
所以你在window平台运行时__main__必须加，避免递归创建进程

spawn 模式会直接将main代码外的代码在子解释器进程中复制一遍

#spawn 会拷贝出main代码外的所需资源
name = 100
def fun1():
    print("xxxx")
    global name
    print(name)
    print(threading.enumerate())
    print(multiprocessing.current_process().pid)

if __name__ == '__main__':
    multiprocessing.set_start_method('spawn')
    process=Process(target=fun1)
    process.start()

fork 全部的资源，且传参数为支持线程锁和文件对象
fork是直接保存一个跟主线程一样状态的资源

#一个有趣的例子:
#正确的输出是: kerwin kurt kerwin
#你想想为什么呢
from multiprocessing import Process
file=open('./name.txt','w+',encoding='utf-8')
file.write("kerwin"+'\n')
def write_info():
    file.write("kurt"+'\n')
    file.flush()
if __name__ == '__main__':
    p1=Process(target=write_info)
    p1.start()
    p1.join()
    file.close()

forkserver和spawn都是直接拷贝很多资源，所以会出现递归，需要使用__main__的问题

###关于进程锁和进程之间资源进行共享

1.对于进程之间的数据共享我们一般都是使用像第三方redis这样的数据来通信
我下面介绍一下原生的交互
shared memory
1.利用python底部的c语言代码，类比于c语言的指针 使用Value和 Array

from multiprocessing import Process,Value,Array


def func():
    num1.value=100
#int num1=123
num1=Value('i',123)
char1=Value('c',b'b')
p1=Process(target=func)
p1.start()
p1.join()
print('main',num1)

2.利用一个manager对象

from multiprocessing import Process,Manager
def fun1():
    li.append('123')

manager=Manager()
li=manager.list([1,2,4])
p1=Process(target=fun1)
p1.start()
p1.join()
print(li)

3.利用Queue管道(单向的)

from multiprocessing import Process,Queue
def fun1():
    queue.put((1,2,3))
    queue.put((1,2,3))


queue=Queue()
p1=Process(target=fun1)
p1.start()
p1.join()
print(queue.get())

4.pipe（双向的）

import multiprocessing
from multiprocessing import Process,Pipe
def fun1():
    c1.send(('passs'))
    print(c1.recv())
c1,q1=multiprocessing.Pipe()
p1=Process(target=fun1)
p1.start()
q1.send("sskdfksf")
p1.join()
print(q1.recv())

上面的几种都基本上不用,嘿嘿嘿

###关于进程锁

import multiprocessing
import time
from multiprocessing import Process,Manager,Value

def fun():
    lock.acquire()
    num.value=num.value-1
    lock.release()
with Manager() as manager:
    #进程都共用一个lock
    num=Value('I',100)
    lock=manager.RLock()
    for i in range(100):
        p=Process(target=fun)
        p.start()
    time.sleep(20)
    print(num.value)

或者

from multiprocessing import Process,Manager,Value

def fun():
    lock.acquire()
    num.value=num.value-1
    print(num.value)
    lock.release()
    #进程都共用一个lock
num=Value('I',100)
lock=multiprocessing.Lock()
for i in range(100):
    p=Process(target=fun)
    p.start()

import multiprocessing
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Manager


def func(lock):
    with lock:
        print('enter',lock)
        print(multiprocessing.current_process().pid)
        with open("./num.txt",'r',encoding='UTF-8') as fp:
            count=int(fp.read())
        count-=1
        with open('./num.txt','w',encoding='UTF-8') as fp:
            fp.write(str(count))
        print("unlocked")

关于协程见另一篇文章