python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > python多线程比单线程效率低的原因

python多线程比单线程效率低的原因及其解决方案

作者：易爻64

这篇文章主要介绍了python多线程比单线程效率低的原因及其解决方案,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教

python多线程比单线程效率低的原因

Python语言的标准实现叫作CPython，它分两步来运行Python程序

步骤1：解析源代码文本，并将其编译成字节码（bytecode）

步骤2：CPython采用基于栈的解释器来运行字节码。

GIL

GIL实际上就是一种互斥锁（mutual-exclusion lock，mutex），用来防止CPython的状态在抢占式的多线程环境（preemptive multithreading）之中受到干扰，因为在这种环境下，一条线程有可能突然打断另一条线程抢占程序的控制权。如果这种抢占行为来得不是时候，那么解释器的状态（例如为垃圾回收工作而设立的引用计数等）就会遭到破坏。

CPython要通过GIL阻止这样的动作，以确保它自身以及它的那些C扩展模块能够正确地执行每一条字节码指令。

GIL会产生一个很不好的影响。在C++与Java这样的语言里面，如果程序之中有多个线程能够分头执行任务，那么就可以把CPU的各个核心充分地利用起来。尽管Python也支持多线程，但这些线程受GIL约束，所以每次或许只能有一条线程向前推进，而无法实现多头并进。

所以，想通过多线程做并行计算或是给程序提速的开发者，恐怕要失望了。

多线程下的线程执行

多线程效率低于单线程原因

如上我们可以知道，在 python中想要某个线程要执行必须先拿到 GIL这把锁，且 python只有一个 GIL，拿到这个 GIL才能进入 CPU执行， 在遇到 I/O操作时会释放这把锁。如果是纯计算的程序，没有 I/O 操作，解释器会每隔 100次操作就释放这把锁，让别的线程有机会 执行（这个次数可以通sys.setcheckinterval来调整）。所以虽然 CPython 的线程库直接封装操作系统的原生线程，但 CPython 进程做为一个整体，同一时间只会有一个获得了 GIL 的线程在跑，其它的线程都处于等待状态等着 GIL 的释放。

而每次释放 GIL锁，线程进行锁竞争、切换线程，会消耗资源。并且由于 GIL锁存在，python里一个进程永远只能同时执行一个线程 (拿到 GIL的线程才能执行 )，这就是为什么在多核 CPU上， python的多线程效率并不高

多线程效率低于或高于单线程原因

相同的代码，为何有时候多线程会比单线程慢，有时又会比单线程快？ 这主要跟运行的代码有关：

CPU密集型代码(各种循环处理、计数等等 )，在这种情况下，由于计算工作多， ticks计数很快就会达到 100阈值，然后触发 GIL的释放与再竞争 （多个线程来回切换当然是需要消耗资源的），所以 python下的多线程遇到 CPU密集型代码时，单线程比多线程效率高。

IO密集型代码 (文件处理、网络爬虫等 )，多线程能够有效提升效率单线程下有 IO操作会进行 IO等待，造成不必要的时间浪费。开启多线程能在线程 A等待时，自动切换到线程 B，可以不浪费 CPU的资源，从而能提升程序执行效率 。进行IO密集型的时候可以进行分时切换 所有这个时候多线程快过单线程

如果python想充分利用多核 CPU，可以采用多进程

每个进程有各自独立的 GIL，互不干扰，这样就可以真正意义上的并行执行。

在 python中，多进程的执行效率优于多线程 (仅仅针对多核 CPU而言 )。所以在多核 CPU下，想做并行提升效率，比较通用的方法是使用多进程，能够有效提高执行效率

代码示例：

# 多线程
# 最后完成的线程的耗时
# [TIME MEASURE] execute function: gene_1000_field took 3840.604ms
@time_measure
def mult_thread(rows):
    # 总行数
    rows = rows
    # 线程数
    batch_size = 4
    cell = math.ceil(rows / batch_size)
    # 处理数据生成
    print('数据生成中，线程数:' + str(batch_size))
    threads = []
    for i in range(batch_size):
        starts = i * cell
        ends = (i + 1) * cell
        file = f"my_data_{str(i)}.csv"
        # t = threading.Thread(target=gene_1000_field_test, args=(starts, ends, file))
        t = threading.Thread(target=gene_1000_field, args=(starts, ends, file))
        t.start()
        threads.append(t)
    # for t in threads:
    #     t.join()
# 多进程
# [TIME MEASURE] execute function: gene_1000_field took 1094.776ms
# 执行时间和单个线程的执行时间差不多，目的达到
@time_measure
def mult_process(rows):
    # 总行数
    rows = rows
    # 线程数
    batch_size = 4
    cell = math.ceil(rows / batch_size)
    # 处理数据生成
    print('数据生成中，线程数:' + str(batch_size))
    process = []
    for i in range(batch_size):
        starts = i * cell
        ends = (i + 1) * cell
        file = f"my_data_{str(i)}.csv"
        # p = Process(target=f, args=('bob',))
        # p.start()
        # p_lst.append(p)
        # t = threading.Thread(target=gene_1000_field_test, args=(starts, ends, file))
        p = Process(target=gene_1000_field, args=(starts, ends, file))
        p.start()
        process.append(p)

python中多线程与单线程的对比

# 做一个简单的爬虫：
import threading
import time
import functools
from urllib.request import urlopen
# 写一个时间函数的装饰器
def timeit(f):
    @functools.wraps(f)
    def wrapper(*args,**kwargs):
        start_time=time.time()
        res=f(*args,**kwargs)
        end_time=time.time()
        print("%s函数运行时间:%.2f" % (f.__name__, end_time - start_time))
        return res
    return wrapper
def get_addr(ip):
    url="http://ip-api.com/json/%s"%(ip)
    urlobj=urlopen(url)
    # 服务端返回的页面信息, 此处为字符串类型
    pagecontent=urlobj.read().decode('utf-8')
    # 2. 处理Json数据
    import json
    # 解码: 将json数据格式解码为python可以识别的对象;
    dict_data = json.loads(pagecontent)
    print("""
    ip : %s
    所在城市: %s
    所在国家: %s
    """ % (ip, dict_data['city'], dict_data['country']))
#不使用多线程
@timeit
def main1():
    ips = ['12.13.14.%s' % (i + 1) for i in range(10)]
    for ip in ips:
        get_addr(ip)
# 多线程的方法一
@timeit
def main2():
    ips=['12.13.14.%s'%(i+1) for i in range(10)]
    threads=[]
    for ip in ips:
        t=threading.Thread(target=get_addr,args=(ip,))
        threads.append(t)
        t.start()
    [thread.join() for thread in threads]
# 多线程的方法二
class MyThread(threading.Thread):
    def __init__(self, ip):
        super(MyThread, self).__init__()
        self.ip = ip
    def run(self):
        url = "http://ip-api.com/json/%s" % (self.ip)
        urlObj = urlopen(url)
        # 服务端返回的页面信息, 此处为字符串类型
        pageContent = urlObj.read().decode('utf-8')
        # 2. 处理Json数据
        import json
        # 解码: 将json数据格式解码为python可以识别的对象;
        dict_data = json.loads(pageContent)
        print("""
                            %s
        所在城市: %s
        所在国家: %s
        """ % (self.ip, dict_data['city'], dict_data['country']))
@timeit
def main3():
    ips = ['12.13.14.%s' % (i + 1) for i in range(10)]
    threads = []
    for ip in ips:
        t = MyThread(ip)
        threads.append(t)
        t.start()
    [thread.join() for thread in threads]
if __name__ == '__main__':
    main1()
    main2()
    main3()

---->输出：
# main1函数运行时间:55.06
# main2函数运行时间:5.64
# main3函数运行时间:11.06

由次可以看出多线程确实速度快了很多，然而这只是适合I/O密集型，当计算密集型中cpu一直在占用的时候，多线程反而更慢。

下面举例

import threading
import time
def my_counter():
    i = 1
    for count in range(200000000):
        i = i + 2*count
    return True
# 采用单线程
@timeit
def main1():
    thread_array = {}
    for tid in range(2):
        t = threading.Thread(target=my_counter)
        t.start()
        t.join()
# 采用多线程
@timeit
def main2():
    thread_array = {}
    for tid in range(2):
        t = threading.Thread(target=my_counter)
        t.start()
        thread_array[tid] = t
    for i in range(2):
        thread_array[i].join()
if __name__ == '__main__':
    main1()
    main2()

----->输出：
main1函数运行时间:27.57
main2函数运行时间:28.19

这个时候就能体现出来多线程适应的场景

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:
阅读全文