-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy paththreading.Timer.py
150 lines (122 loc) · 6.4 KB
/
threading.Timer.py
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
####################################################################################
'''
Класс Timer() модуля threading в Python.
Создание и запуск потоков-таймеров.
Синтаксис:
import threading
timer = threading.Timer(interval, function, args=None, kwargs=None)
Параметры:
interval - интервал запуска вызываемого объекта (функции),
function - вызываемый объект (функция),
args=None - позиционные аргументы function,
kwargs=None - ключевые аргументы function.
Возвращаемое значение:
объект Timer.
Описание:
Класс Timer() модуля threading создает таймер, который будет запускать функцию function
с аргументами args и ключевыми аргументами kwargs по прошествии интервала interval секунд.
Если args равен None (по умолчанию), то будет использоваться пустой список.
Если kwargs равен None (по умолчанию), то будет использоваться пустой словарь.
Этот класс представляет действие, которое следует запускать только по прошествии
определенного времени - таймер. Таймер является подклассом threading.Thread() и,
как таковой, также служит примером создания пользовательских потоков.
Таймеры запускаются, как и потоки, путем вызова их метода Timer.start(),
унаследованного от класса threading.Thread().
Таймер можно остановить до того, как его действие начнется, вызвав метод Timer.cancel().
Интервал, который таймер будет ожидать перед выполнением своего действия,
может не совпадать с интервалом, указанным пользователем.
'''
####################################################################################
# через 30 секунд будет напечатано "hello, world".
from threading import Timer
def hello():
print("hello, world")
t = Timer(30.0, hello)
t.start()
####################################################################################
'''
Методы объекта Timer.
Объект Timer дополнительно определяет один метод.
Остальные методы он наследует от класса threading.Thread().
Timer.cancel():
Метод Timer.cancel() останавливает таймер и отменяет выполнение действия таймера.
Метод будет работать только в том случае, если таймер все еще находится
в стадии ожидания. Сработавший таймер остановить нельзя.
'''
####################################################################################
'''
Примеры создания и запуска потоков-таймеров.
Таймер начинает свою работу после задержки и может быть отменен в любой момент
в течение этого времени задержки.
Второй таймер в этом примере никогда не запускается, а первый таймер запускается
после того, как остальная часть основной программы завершена.
Так как первый таймер это не поток демона, то он присоединяется неявно,
когда основной поток завершится, по этому delayed() сообщение не напечатает.
'''
import threading, time
def delayed():
th_name = threading.current_thread().name
print(f'Th:{th_name} Worker запущен')
# Создание и запуск потоков таймеров
t1 = threading.Timer(0.3, delayed)
t1.name = 'Timer-1'
t2 = threading.Timer(0.3, delayed)
t2.name = 'Timer-2'
print('Запуск таймеров')
t1.start()
t2.start()
print(f'Ожидание перед завершением {t2.name}')
time.sleep(0.2)
print(f'Завершение {t2.name}')
t2.cancel()
print('Выполнено')
# Запуск таймеров
# Ожидание перед завершением Timer-2
# Завершение Timer-2
# Выполнено
# Th:Timer-1 Worker запущен
####################################################################################
'''
Пример выхода из программы после двух секунд (первая версия).
'''
class ExamTimer:
def __init__(self, n: int) -> None:
from threading import Timer
self.t = Timer(n, self.exit)
self.t.start()
def exit(self):
print('Алгоритм не успел решить задачу по времени')
__import__('os').abort()
if __name__ == '__main__':
ExamTimer(2)
for i in range(10000000000000000000000000000):
print(i)
####################################################################################
'''
Пример выхода из программы после двух секунд (вторая версия).
'''
class ExamTimer:
def __init__(self, n: int) -> None:
from threading import Timer
self.t = Timer(n, self.exit)
def cancel(self):
self.t.cancel()
def exit(self):
print('Алгоритм не успел решить задачу по времени')
__import__('os').abort()
def start(self):
self.t.start()
class OdinDvaTri:
def __init__(self, n: int) -> None:
self.my_list = range(1, n + 1)
self.n = n
def algo(self) -> str:
timer = ExamTimer(2)
timer.start()
for i in range(10000000000000000000000000000):
print(i)
timer.cancel()
return 'Done'
if __name__ == '__main__':
print(OdinDvaTri(int(input())).algo())
####################################################################################