forked from Jayhello/python_utils
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathpool_dummy.py
More file actions
56 lines (43 loc) · 1.39 KB
/
Copy pathpool_dummy.py
File metadata and controls
56 lines (43 loc) · 1.39 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
# _*_ coding:utf-8 _*_
"""
This file is about thread(dummy)/process pool
"""
from multiprocessing import Pool as ProcessPool
from multiprocessing.dummy import Pool as ThreadPool
import logging
from time import sleep, time
from random import randrange
# logging.basicConfig(level=logging.DEBUG,
# format='%(levelname)s %(asctime)s %(threadName)s %(message)s',
# datefmt='%Y-%m-%d %I:%M:%S')
logging.basicConfig(level=logging.DEBUG,
format='%(levelname)s %(asctime)s %(processName)s %(message)s',
datefmt='%Y-%m-%d %I:%M:%S')
def handler(sec):
logging.debug('now I will sleep %s S', sec)
sleep(sec)
def get_pool(b_dummy=True, num=4):
"""
if b_dummy is True then get ThreadPool, or get process pool
:param b_dummy: dummy thread Pool or Process pool
:param num: thread or process num
:return: pool object
"""
if b_dummy:
pool = ThreadPool(num)
else:
pool = ProcessPool(num)
return pool
def test_dummy_thread_pool():
start_time = time()
lst_sleep_sec = [randrange(3, 10) for i in xrange(10)]
pool = get_pool(b_dummy=False)
results = pool.map(handler, lst_sleep_sec)
logging.debug(results)
pool.close()
pool.join()
logging.debug('time consume %s', time() - start_time)
pass
if __name__ == '__main__':
test_dummy_thread_pool()
pass