test_connection.py 文件源码

def test_with_multi_threading():
    test_truncate()

    def task(n):
        print('In thread {}'.format(threading.get_ident()))
        for _ in range(n):
            test_insert_one()

    threads = [threading.Thread(target=task, args=(100,)) for _ in range(50)]
    for t in threads:
        t.start()

    for t in threads:
        t.join()

    test_query()