forked from Jayhello/python_utils
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy paththread_condition.py
More file actions
45 lines (35 loc) · 1.13 KB
/
Copy paththread_condition.py
File metadata and controls
45 lines (35 loc) · 1.13 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
# _*_ coding:utf-8 _*_
"""
This file is about a sample demo about threading.condition
"""
import threading
import time
import logging
from basic_thread import join_all_others_thread
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(threadName)s %(message)s',
datefmt='%Y-%m-%d %I:%M:%S')
def notify_condition(con):
with con:
logging.debug('now notify all the condition')
con.notifyAll()
def wait_condition(con):
with con:
logging.debug('I am waiting for an condition')
con.wait()
logging.debug('I get the condition.....')
def test_demo():
con = threading.Condition()
t_w1 = threading.Thread(name='t_w1', target=wait_condition, args=(con, ))
t_w2 = threading.Thread(name='t_w2', target=wait_condition, args=(con, ))
t_n1 = threading.Thread(name='t_n1', target=notify_condition, args=(con, ))
t_w1.start()
t_w2.start()
logging.debug('now main thread sleeping 5S')
time.sleep(5)
t_n1.start()
join_all_others_thread()
logging.debug('now all have been done')
if __name__ == '__main__':
test_demo()
pass