多线程和多进程1

一. GIL

 

1. 基本概念

GIL：global interpreter lock （cpython下）

python中一个线程对应于c语言中的一个线程，GIL使得同一时刻只有一个线程在一个CPU上执行字节码，无法将多个线程映射到多个CPU上执行

 

 

2. GIL释放

1）GIL会根据执行的字节码行数以及时间片释放GIL

2）GIL在遇到I/O操作时会主动释放

 

total = 0

def add():

    global total

    for i in range(1000000):

        total += 1

 

def desc():

    global total

    for i in range(1000000):

        total -= 1

 

import threading

thread1 = threading.Thread(target=add)

thread2 = threading.Thread(target=desc)

thread1.start()

thread2.start()

 

thread1.join()

thread2.join()

print(total)

 

按理说，最后的结果应该是0，但是多次运行程序得到的值不是0，且每次都是不一样的，证明并非add函数执行完后，才会执行desc函数，也就是在函数add执行期间释放了GIL，去执行了desc函数

 

 

3. 下面用一个小例子用字节码来解释为什么上面结果不为0

import dis

 

def add(a):

    a += 1

 

def desc(a):

    a -= 1

 

print(dis.dis(add))

print(dis.dis(desc))

输出结果

  4           0 LOAD_FAST                0 (a)                  

              2 LOAD_CONST               1 (1)                

              4 INPLACE_ADD                                        

              6 STORE_FAST               0 (a)                     

              8 LOAD_CONST               0 (None)

             10 RETURN_VALUE

None

  7           0 LOAD_FAST                0 (a)

              2 LOAD_CONST               1 (1)

              4 INPLACE_SUBTRACT

              6 STORE_FAST               0 (a)

              8 LOAD_CONST               0 (None)

             10 RETURN_VALUE

 

说明：

add函数字节码前4步， desc类似

1）load a

2）load 1

3）+操作

4）赋值给a

 

add的字节码和desc的字节码是并行执行的，且全局变量是共用的，所以两个线程的加法和减法操作使的变量a一直变化。GIL释放的时候可能是得到add的结果，也可能是得到desc的结果. 最终的返回值是在两个值间摇摆的

 

 

 

 

 

二. 多线程编程-threading

 

对于I/O操作来说，多线程和多进程性能差别不大

 

1. 通过Thread类实例化，实现多线程编程

import time

import threading

 

def eat(x):

    print("start eat")

    time.sleep(2)

    print("end eat")

 

def drink(x):

    print("start drink")

    time.sleep(2)

    print("end drink")

 

if __name__ == "__main__":

    thread1 = threading.Thread(target=eat, args=("",))

    thread2 = threading.Thread(target=drink, args=("",))

    start_time = time.time()

    thread1.start()  #启动线程

    thread2.start()

 

    print("last time: {}".format(time.time()-start_time))

    

输出结果如下

start eat

start drink

last time: 0.0009996891021728516

end drink

end eat

说明：

1）结果显示持续时间不是2，而是接近于0，原因是程序运行print语句时自动生成一个主线程，它和自定义的2个线程是并行的，不需要等待自定义的2个线程结束后才开始执行

2）主线程的print语句执行结束后，会接着执行thread1, thread2，2秒后打印出"end drink", "end eat"; 如果想print语句结束后，自定义的2个线程也立刻结束，可以在定义thread2变量后加下面两条代码

thread1.setDaemon(True)  #把thread1变为守护线程

thread2.setDaemon(True)

结果输出

start eat

start drink

last time: 0.0009982585906982422

可以自己测试一下只设置一个为守护线程，并修改sleep时间后的输出

 

3）如果想先执行thread1, thread2结束后，再执行print语句的主线程，修改代码如下

import time

import threading

 

def eat(x):

    print("start eat")

    time.sleep(2)

    print("end eat")

 

def drink(x):

    print("start drink")

    time.sleep(4)

    print("end drink")

 

if __name__ == "__main__":

    thread1 = threading.Thread(target=eat, args=("",))

    thread2 = threading.Thread(target=drink, args=("",))

    #thread1.setDaemon(True)  这里注释掉，方便测试

    #thread2.setDaemon(True)

    start_time = time.time()

    thread1.start()  #启动线程

    thread2.start()

 

    thread1.join() 

    thread2.join()

 

    print("last time: {}".format(time.time()-start_time))

输出结果

start eat

start drink

end eat

end drink

last time: 4.001539468765259

运行时间为4， 说明是按照线程执行之间最长的计算，而不是2个线程的执行时间总和

 

 

 

2. 通过继承Thread来实现多线程

import time

import threading

 

 

class Eat(threading.Thread):

    # 给线程命名

    def __init__(self, name):

        super().__init__(name=name)

    # 重载run()

    def run(self):

        print("start eat")

        time.sleep(2)

        print("end eat")

 

 

class Drink(threading.Thread):

    def __init__(self, name):

        super().__init__(name=name)

    # 重载run()

    def run(self):

        print("start drink")

        time.sleep(3)

        print("end drink")

 

if __name__ == "__main__":

    thread1 = Eat("chirou")

    thread2 = Drink("hejiu")

    start_time = time.time()

    thread1.start()  #启动线程

    thread2.start()

 

    thread1.join()

    thread2.join()

 

    print("last time: {}".format(time.time()-start_time))

输出

start eat

start drink

end eat

end drink

last time: 3.001695394515991

 

 

 

 

 

 

三. 线程间通信-共享变量和Queue

 

1. 共享变量--定义全局变量来让两个线程共用

import time

import threading

 

detail_url_list = []

 

def get_detail_html():

    # 爬取文章详情页

    global detail_url_list

    if len(detail_url_list):

        url = detail_url_list.pop()

        print("get detail html started")

        time.sleep(2)

        print("get detail html end")

 

 

def get_detail_url():

    # 爬取文章列表页，然后交给详情页函数

    global detail_url_list

    print("get detail url started")

    time.sleep(4)

    for i in range(20):

        detail_url_list.append("http://projectsedu.com/{id}".format(id=i))

    print("get detail url end")

 

 

if __name__ == "__main__":

    thread_detail_url = threading.Thread(target=get_detail_url)

    for i in range(3):

        html_thread = threading.Thread(target=get_detail_html)

        html_thread.start()

 

    start_time = time.time()

    print("last time: {}".format(time.time()-start_time))

问题：线程不安全，不同线程中可能会影响变量值，需要添加锁

 

 

 

2. 使用queue的方式进行线程间同步，队列是线程安全的

from queue import Queue

import time

import threading

 

 

def get_detail_html(queue):

    #爬取文章详情页

    while True:

        url = queue.get() #从队列中取数据，如果队列为空会一直停在这一行

        print("get detail html started")

        time.sleep(2)

        print("get detail html end")

 

 

def get_detail_url(queue):

    # 爬取文章列表页

    while True:

        print("get detail url started")

        time.sleep(4)

        for i in range(20):

            queue.put("http://projectsedu.com/{id}".format(id=i))  #队列里放数据

        print("get detail url end")

 

 

if __name__ == "__main__":

    detail_url_queue = Queue(maxsize=1000) #设置队列最大值

    thread_detail_url = threading.Thread(target=get_detail_url, args=(detail_url_queue,))

    for i in range(10):

        html_thread = threading.Thread(target=get_detail_html, args=(detail_url_queue,))

        html_thread.start()

 

    start_time = time.time()

    print("last time: {}".format(time.time()-start_time))

 

Queue类的几个函数介绍

full()：判断队列是否已满

qsize(): 返回队列大小

empty(): 判断队列是否为空

join(): 使队列处于阻塞状态，只有接收到task_done()时，join()函数才会退出。所以这两个函数是成对出现的

 

 

 

 

 

四. 线程同步--Lock, RLock

 

1. Lock：给代码段加锁，此时程序只能执行锁中的代码。只有锁释放后，才能执行另外一段代码

改写第一节中GIL释放问题的代码，加上锁，结果就是0了

from threading import Lock

 

total = 0

lock = Lock()

 

def add():

    global total

    global lock

    for i in range(1000000):

        lock.acquire() #获取锁

        total += 1

        lock.release() #释放锁

 

def desc():

    global total

    global lock

    for i in range(1000000):

        lock.acquire()

        #lock.acquire()   死锁情况1：连续2次使用lock.acquire()，就会造成死锁，程序一直不执行

        total -= 1

        lock.release()

 

import threading

thread1 = threading.Thread(target=add)

thread2 = threading.Thread(target=desc)

thread1.start()

thread2.start()

 

thread1.join()

thread2.join()

print(total)

 

注意：

1）获取锁和释放锁都需要时间，所以锁会影响性能

2）锁会引起死锁，死锁情况2如下

A(a, b)

acquire(a)  #需要先获得a，然后获得b

acquire(b)

B(a, b)

acquire(b)  #需要先获得b, 然后获得a

acquire(a)

如果A(a, b)获得a的同时，B(a, b)获得了b，那么他们都在互相等待资源造成死锁

 

 

2. RLock: 可重入锁

同一个线程里面可调用多次acquire(), 解决某函数参数为函数，并且也有lock的情况

from threading import Lock, RLock

 

total = 0

lock = RLock()

 

def add():

    global total

    global lock

    for i in range(1000000):

        lock.acquire()  

        lock.acquire()

        total += 1

        lock.release()

        lock.release()

 

def desc():

    global total

    global lock

    for i in range(1000000):

        lock.acquire()

        total -= 1

        lock.release()

 

import threading

thread1 = threading.Thread(target=add)

thread2 = threading.Thread(target=desc)

thread1.start()

thread2.start()

 

thread1.join()

thread2.join()

print(total)

add函数中加锁和释放锁的次数要一样

 

 

 

 

 

四. 线程同步--condition使用以及源码分析

 

condition: 条件变量，用于复杂的线程间同步，比如模拟机器人对话

import threading

 

class XiaoAi(threading.Thread):

    def __init__(self, cond):

        super().__init__(name="小爱")

        self.cond = cond

 

    def run(self):

        with self.cond:

            self.cond.wait()

            print("{} : 在 ".format(self.name))

            self.cond.notify()

 

            self.cond.wait()

            print("{} : 好啊 ".format(self.name))

            self.cond.notify()

 

 

class TianMao(threading.Thread):

    def __init__(self, cond):

        super().__init__(name="天猫精灵")

        self.cond = cond

 

    def run(self):

        with self.cond:

            print("{} : 小爱同学 ".format(self.name))

            self.cond.notify()

            self.cond.wait()

 

            print("{} : 我们来对古诗吧 ".format(self.name))

            self.cond.notify()

            self.cond.wait()

 

 

if __name__ == "__main__":

    from concurrent import futures

    cond = threading.Condition()

    xiaoai = XiaoAi(cond)

    tianmao = TianMao(cond)

 

    #启动顺序很重要

    #在调用with cond之后才能调用wait或者notify方法

    #condition有两层锁， 一把底层锁会在线程调用了wait方法的时候释放， 上面的锁会在每次调用wait的时候分配一把并放入到cond的等待队列中，等到notify方法的唤醒

    xiaoai.start()

    tianmao.start()

输出

天猫精灵 : 小爱同学

小爱 : 在

天猫精灵 : 我们来对古诗吧

小爱 : 好啊

 

说明：

1）线程的启动顺序很重要，如果先执行tianmao.start()，再启动xiaoai.start()，程序就会卡在"天猫精灵：小爱同学"这里，原因是wait()必须要有notify()通知后才能响应。当tianmao线程启动发出notify时，xiaoai线程还没启动，所以里面的wait()一直不能得到响应

 

2）可以看一下queue的源码来理解conditon的应用

from queue import Queue