任务

def single():

    # 单进程单线程实现

    s = 0

    for i in range(1, N):

        s += math.sqrt(i)

    return s

结论

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

实现

import math

import multiprocessing

import threading

import timeit

import numpy as np

from twisted.internet import reactor

import time

N = 10000000

def single():

    # 单进程单线程实现

    s = 0

    for i in range(1, N):

        s += math.sqrt(i)

    return s

def useThread():

    # 多线程实现

    total_sum = 0

    def go(beg, end):

        nonlocal total_sum

        s = 0

        for i in range(beg, end):

            s += math.sqrt(i)

        total_sum += s  # python无法利用多核，所以这句话每个时刻只有一个线程在执行

    thread_count = 4

    per = 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_sum

def useMultiprocess():

    # 使用多进程

    def go(q: multiprocessing.Queue, beg, end):

        s = 0

        for i in range(beg, end):

            s += math.sqrt(i)

        q.put(s)

    process_count = 4

    per = 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 = 0

    try:

        while 1:

            x = q.get_nowait()

            total_sum += x

    except:

        pass

    return total_sum

def useTwisted():

    # reactor是单例模式,一个进程只有一个reactor,一个reactor包括多个线程

    total_sum = 0

    ok_count = 0

    thread_count = 4

    def go(beg, end):

        nonlocal total_sum

        s = 0

        for i in range(beg, end):

            s += math.sqrt(i)

        reactor.callFromThread(accumulate, s)

    def accumulate(s):

        nonlocal total_sum

        nonlocal ok_count

        ok_count += 1

        if ok_count == thread_count:

            reactor.stop()

        total_sum += s

    def 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 = 0

    thread_count = 4

    ok_count = 0

    beg_time = time.time()

    def go(beg, end):

        nonlocal total_sum

        s = 0

        for i in range(beg, end):

            s += math.sqrt(i)

        reactor.callFromThread(accumulate, s)

    def accumulate(s):

        nonlocal total_sum

        nonlocal ok_count

        total_sum += s

        ok_count += 1

        if 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 math

beg_time = time.time()

def go():

    print("go start")

    s = 0

    for i in range(10000000):

        s += math.sqrt(i + 1)

    print("go over", time.time() - beg_time)

import timeit

reactor.suggestThreadPoolSize(8)

print(timeit.timeit(go, number=1))

for i in range(10):

    reactor.callInThread(go)

reactor.run()