任务

def single():# 单进程单线程实现s = 0for i in range(1, N):s += math.sqrt(i)return s

结论

• Python多线程无法利用多核
• Python多进程可以利用多核
• Numpy速度远超并行的Python代码
• twisted无法利用多核

实现

import math
import multiprocessing
import threading
import timeitimport numpy as np
from twisted.internet import reactor
import timeN = 10000000def single():# 单进程单线程实现s = 0for i in range(1, N):s += math.sqrt(i)return sdef useThread():# 多线程实现total_sum = 0def go(beg, end):nonlocal total_sums = 0for i in range(beg, end):s += math.sqrt(i)total_sum += s  # python无法利用多核，所以这句话每个时刻只有一个线程在执行thread_count = 4per = math.ceil(N / thread_count)thread_list = []for i in range(thread_count):th = threading.Thread(target=go, args=(i * per, (i + 1) * per))thread_list.append(th)th.start()for th in thread_list:th.join()return total_sumdef useMultiprocess():# 使用多进程def go(q: multiprocessing.Queue, beg, end):s = 0for i in range(beg, end):s += math.sqrt(i)q.put(s)process_count = 4per = math.ceil(N / process_count)process_list = []q = multiprocessing.Queue()for i in range(process_count):th = multiprocessing.Process(target=go, args=(q, i * per, (i + 1) * per))process_list.append(th)th.start()for th in process_list:th.join()total_sum = 0try:while 1:x = q.get_nowait()total_sum += xexcept:passreturn total_sumdef useTwisted():# reactor是单例模式,一个进程只有一个reactor,一个reactor包括多个线程total_sum = 0ok_count = 0thread_count = 4def go(beg, end):nonlocal total_sums = 0for i in range(beg, end):s += math.sqrt(i)reactor.callFromThread(accumulate, s)def accumulate(s):nonlocal total_sumnonlocal ok_countok_count += 1if ok_count == thread_count:reactor.stop()total_sum += sdef process_work(q):reactor.suggestThreadPoolSize(thread_count)per = math.ceil(N / thread_count)for i in range(thread_count):reactor.callInThread(go, i * per, i * per + per)reactor.run()q.put(total_sum)q = multiprocessing.Queue()p = multiprocessing.Process(target=process_work, args=(q,))p.start()p.join()return q.get()def useTwisted2():# reactor是单例模式,一个进程只有一个reactor,一个reactor包括一个线程total_sum = 0thread_count = 4ok_count = 0beg_time = time.time()def go(beg, end):nonlocal total_sums = 0for i in range(beg, end):s += math.sqrt(i)reactor.callFromThread(accumulate, s)def accumulate(s):nonlocal total_sumnonlocal ok_counttotal_sum += sok_count += 1if ok_count == thread_count:print(time.time() - beg_time, "value", total_sum)reactor.suggestThreadPoolSize(thread_count)per = math.ceil(N / thread_count)for i in range(thread_count):reactor.callInThread(go, i * per, i * per + per)def useNumpy():a = np.linspace(1, N, N)return np.sum(np.sqrt(a))def main():for method in (single, useThread, useMultiprocess, useNumpy, useTwisted, useTwisted2):print(method.__name__, "result", method(), "time", timeit.timeit(method, number=10))reactor.run()if __name__ == '__main__':main()

twisted无法利用多核

from twisted.internet import threads, reactor
import time
import mathbeg_time = time.time()def go():print("go start")s = 0for i in range(10000000):s += math.sqrt(i + 1)print("go over", time.time() - beg_time)import timeitreactor.suggestThreadPoolSize(8)
print(timeit.timeit(go, number=1))
for i in range(10):reactor.callInThread(go)
reactor.run()