python concurrency models compare

bench_concurrency_models.py

import random
import time
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED

from memory_profiler import profile

random.seed(10)
total_iterations = 1000


def generate_a_number(i):
    sleep_time = random.random()
    time.sleep(sleep_time)
    # print("sleep", sleep_time)
    return i


def handle(i):
    print(i)


@profile
def map_all_multi_threading():
    # 有序，性能好，耗内存
    with ThreadPoolExecutor(max_workers=20) as pool:
        results = pool.map(generate_a_number, range(total_iterations))
        for i in results:
            handle(i)


@profile
def one_submit_batch_wait_multi_threading():
    # 有序，性能差，不耗内存
    batch_size = 20
    batch = []
    with ThreadPoolExecutor(max_workers=20) as pool:
        for i in range(total_iterations):
            batch.append(pool.submit(generate_a_number, i))
            if len(batch) >= batch_size:
                for future in batch:
                    handle(future.result())
                batch = []


@profile
def batch_submit_batch_wait_multi_threading():
    # 有序，性能差，不耗内存
    batch_size = 20
    batch = []
    with ThreadPoolExecutor(max_workers=20) as pool:
        for i in range(total_iterations):
            batch.append(i)
            if len(batch) >= batch_size:
                results = pool.map(generate_a_number, batch)
                for result in results:
                    handle(result)
                batch = []


@profile
def windowed_multi_threading():
    # 无序，性能好，不耗内存
    tasks = set()
    max_workers = 20
    generator = (i for i in range(total_iterations))
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        for i in range(max_workers):
            next_task = next(generator, False)
            if next_task:
                tasks.add(pool.submit(generate_a_number, next_task))
        while len(tasks) > 0:
            done_tasks, running_tasks = wait(tasks, return_when=FIRST_COMPLETED)
            for j in done_tasks:
                handle(j.result())
                tasks.remove(j)
                next_task = next(generator, False)
                if next_task:
                    tasks.add(pool.submit(generate_a_number, next_task))


if __name__ == '__main__':
    start_time = time.time()
    # map_all_multi_threading()
    # windowed_multi_threading()
    one_submit_batch_wait_multi_threading()
    # batch_submit_batch_wait_multi_threading()
    print(time.time() - start_time)