manyi/testcase/library/concurrency.py

# -*- coding: utf-8 -*-
#!/usr/bin/env python
"""
taken from django-oscar
用来测试一些并发的情况

"""
try:
    import queue
except ImportError:
    import Queue as queue

import threading

def run_concurrently(fn, kwargs=None, num_threads=5):
    exceptions = queue.Queue()

    def worker(**kwargs):
        try:
            result = fn(**kwargs)
        except Exception as exc:
            exceptions.put(exc)
        else:
            exceptions.put(result)

    kwargs = kwargs if kwargs is not None else {}

    # Run them
    threads = [
        threading.Thread(target=worker, name='thread-%d' % i, kwargs=kwargs)
        for i in range(num_threads)
    ]
    try:
        for thread in threads:
            thread.start()
    finally:
        for thread in threads:
            thread.join()

    # Retrieve exceptions
    exceptions = [exceptions.get(block=False) for _ in range(num_threads)]
    return exceptions