Python多线程 - 墨缘晓生

引言

　　多线程和多进程是不一样的，一个是 thread 库，一个是 multiprocessing 库。GIL的全称是Global Interpreter Lock(全局解释器锁)，来源是python设计之初的考虑，为了数据安全所做的决定。每个CPU在同一时间只能执行一个线程（在单核CPU下的多线程其实都只是并发，不是并行，并发和并行从宏观上来讲都是同时处理多路请求的概念。但并发和并行又有区别，并行是指两个或者多个事件在同一时刻发生；而并发是指两个或多个事件在同一时间间隔内发生。）
　　在Python多线程下，每个线程的执行方式：

获取GIL
执行代码直到sleep或者是python虚拟机将其挂起。
释放GIL 　　可见，某个线程想要执行，必须先拿到GIL，我们可以把GIL看作是“通行证”，并且在一个python进程中，GIL只有一个。拿不到通行证的线程，就不允许进入CPU执行。
在Python2.x里，GIL的释放逻辑是当前线程遇见IO操作或者ticks计数达到100（ticks可以看作是Python自身的一个计数器，专门做用于GIL，每次释放后归零，这个计数可以通过 sys.setcheckinterval 来调整），进行释放。
而每次释放GIL锁，线程进行锁竞争、切换线程，会消耗资源。并且由于GIL锁存在，python里一个进程永远只能同时执行一个线程(拿到GIL的线程才能执行)，这就是为什么在多核CPU上，python的多线程效率并不高。
　　那么是不是python的多线程就完全没用了呢？
1、CPU密集型代码(各种循环处理、计数等等)，在这种情况下，由于计算工作多，ticks计数很快就会达到阈值，然后触发GIL的释放与再竞争（多个线程来回切换当然是需要消耗资源的），所以python下的多线程对CPU密集型代码并不友好。
2、IO密集型代码(文件处理、网络爬虫等)，多线程能够有效提升效率(单线程下有IO操作会进行IO等待，造成不必要的时间浪费，而开启多线程能在线程A等待时，自动切换到线程B，可以不浪费CPU的资源，从而能提升程序执行效率)。所以python的多线程对IO密集型代码比较友好。

　　而在python3.x中，GIL不使用ticks计数，改为使用计时器（执行时间达到阈值后，当前线程释放GIL），这样对CPU密集型程序更加友好，但依然没有解决GIL导致的同一时间只能执行一个线程的问题，所以效率依然不尽如人意。
　　多核性能
　　多核多线程比单核多线程更差，原因是单核下多线程，每次释放GIL，唤醒的那个线程都能获取到GIL锁，所以能够无缝执行，但多核下，CPU0释放GIL后，其他CPU上的线程都会进行竞争，但GIL可能会马上又被CPU0拿到，导致其他几个CPU上被唤醒后的线程会醒着等待到切换时间后又进入待调度状态，这样会造成线程颠簸(thrashing)，导致效率更低

　　多进程为什么不会这样？
　　每个进程有各自独立的GIL，互不干扰，这样就可以真正意义上的并行执行，所以在python中，多进程的执行效率优于多线程(仅仅针对多核CPU而言)。

　　所以在这里说结论：多核下，想做并行提升效率，比较通用的方法是使用多进程，能够有效提高执行效率。
　　Python中使用线程有两种方式：函数或者用类来包装线程对象。

使用函数创建多线程

1	threading._start_new_thread(function,args[,kwargs])

function - 线程函数
args - 传递给线程函数的参数，必须是tuple类型
kwargs - 可选参数

import threading
import time
exitFlag=0
def print_time(threadName,delay):
    count=0
    while count<5:
        time.sleep(delay)
        count+=1
        print('%s:  %s' % (threadName,time.ctime(time.time())))
# 创建两个线程
try:
    threading._start_new_thread(print_time,("Threading_1",2))
    threading._start_new_thread(print_time,("Threading_2",4))
except Exception as e:
    print(e)

　　线程的结束一般依靠线程函数的自然结束；也可以在线程函数中调用thread.exit()，他抛出SystemExit exception，达到退出线程的目的。

使用Threading模块创建线程

　　使用Threading模块创建线程，直接从threading.Thread继承，然后重写__init__方法和run方法：

import threading
import time
exitFlag=0

class MyThread(threading.Thread): # 继承父类threding.Thread
    def __init__(self,threadId,name,counter):
        threading.Thread.__init__(self)
        self.threadId=threadId
        self.name=name
        self.counter=counter
    def run(self):
        # 将要执行的代码写到run函数里面 线程在创建后会直接运行run函数
        print("Starting "+self.name)

def print_time(threadName,delay,counter):
    while counter:
        if exitFlag:
            threadName.exit()
        time.sleep(delay)
        print("%s:  %s" % (threadName,time.ctime(time.time())))
        counter-=1
# 创建新线程
thread1=MyThread(1,"Thread_1",1)
thread2=MyThread(2,"Thread_2",2)

# 开启线程
thread1.start()
thread2.start()

print("Exiting Main Threading")

线程同步

　　如果多个线程共同对某个数据修改，则可能出现不可预料的结果，为了保证数据的正确性，需要对多个线程进行同步。视同Threading的Lock和Rlock可以实现简单的线程同步，这两个对象都有acquire方法和release方法，对于那些需要每次只允许一个线程操作的数据，可以将其操作放到acquire和release方法之间。多线程的优势在于可以同时运行多个任务。但当多线程需要共享数据时，可能存在数据不同步的问题。考虑到这样一种情况：一个列表里所有元素都是0，线程set从后向前把所有元素改成1，而线程print负责从前往后读取列表并打印。那么可能线程set开始的时候，线程print便来打印列表了，输出就成了一半1一半0，这是由于数据不同步造成的。为了避免这种情况，引入锁的概念。锁有两种状态——锁定和未锁定。每当一个线程比如set要访问共享数据时，必须先获得锁定；如果已经有别的线程比如print获得锁定了，那么就让线程set暂停，也就是同步阻塞；等到线程print访问完毕，释放锁以后，再让线程set继续。

import threading
import time


class myThread(threading.Thread):
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter

    def run(self):
        print("Starting " + self.name)

        # 获得锁，成功获得锁定后返回True
        # 可选的timeout参数不填时将一直阻塞直到获得锁定
        # 否则超时后将返回False
        threadLock.acquire()
        print_time(self.name, self.counter, 3)
        # 释放锁
        threadLock.release()


def print_time(threadName, delay, counter):
    while counter:
        time.sleep(delay)
        print("%s: %s" % (threadName, time.ctime(time.time())))
        counter -= 1


threadLock = threading.Lock()
threads = []

# 创建新线程
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# 开启新线程
thread1.start()
thread2.start()

# 添加线程到线程列表
threads.append(thread1)
threads.append(thread2)

# 等待所有线程完成
for t in threads:
    t.join()

　　在上面的代码中运用了线程锁还有join等待。运行结果如下：

Starting Thread-1
Starting Thread-2
Thread-1: Sat Dec 29 20:46:56 2018
Thread-1: Sat Dec 29 20:46:57 2018
Thread-1: Sat Dec 29 20:46:58 2018
Thread-2: Sat Dec 29 20:47:00 2018
Thread-2: Sat Dec 29 20:47:02 2018
Thread-2: Sat Dec 29 20:47:04 2018

线程优先队列

　　Python中Queue模块中提供了同步的、线程安全的队列类，包括FIFO队列Queue，LIFO队列LifoQueue和优先级队列PriorityQueue。这些队列都实现了锁原语，能够在多线程中直接使用。可以直接使用队列来实现线程间的同步。
　　Queue模块中常用的方法：

Queue.qsize() 返回队列的大小
Queue.empty() 如果队列为空，返回True,反之False
Queue.full() 如果队列满了，返回True,反之False
Queue.full 与 maxsize 大小对应
Queue.get([block[, timeout]])获取队列，timeout等待时间
Queue.get_nowait() 相当Queue.get(False)
Queue.put(item) 写入队列，timeout等待时间
Queue.put_nowait(item) 相当Queue.put(item, False)
Queue.task_done() 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号
Queue.join() 实际上意味着等到队列为空，再执行别的操作

#!/usr/bin/python
# -*- coding: UTF-8 -*-
 
import Queue
import threading
import time
 
exitFlag = 0
 
class myThread (threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q
    def run(self):
        print("Starting " + self.name)
        process_data(self.name, self.q)
        print("Exiting " + self.name)
 
def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print "%s processing %s" % (threadName, data)
        else:
            queueLock.release()
        time.sleep(1)
 
threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = Queue.Queue(10)
threads = []
threadID = 1
 
# 创建新线程
for tName in threadList:
    thread = myThread(threadID, tName, workQueue)
    thread.start()
    threads.append(thread)
    threadID += 1
 
# 填充队列
queueLock.acquire()
for word in nameList:
    workQueue.put(word)
queueLock.release()
 
# 等待队列清空
while not workQueue.empty():
    pass
 
# 通知线程是时候退出
exitFlag = 1
 
# 等待所有线程完成
for t in threads:
    t.join()
print("Exiting Main Thread")

输出结果：

Starting Thread-1
Starting Thread-2
Starting Thread-3
Thread-3 processing One
Thread-2 processing Two
Thread-1 processing Three
Thread-3 processing Four
Thread-2 processing Five
Exiting Thread-1
Exiting Thread-3
Exiting Thread-2
Exiting Main Thread

参考:
https://cuiqingcai.com/3325.html

文章目录

引言

使用函数创建多线程

使用Threading模块创建线程

线程同步

线程优先队列