上一篇文章介绍了 Python 程序在 Windows 系统中监测键盘输入的两种方法,即 keyboard 和 pynput 工具包,但在远程控制 Linux 服务器时,这两种方法都会遇到问题: Python: Windows 系统中监测键盘输入的两种方法
本文介绍通过开启新线程监测事件的方式,实现程序运行中的键盘输入监测,这种方法在远程控制 Linux 服务器时也可正常使用。
首先,定义一个可以主动控制其结束的线程,需要继承 threading 中的 Thread 类,并重写其 _stop 方法:
import threading
class StoppableThread(threading.Thread):
"""
a stoppable thread, inherit from threading.Thread
"""
def __init__(self, target):
super(StoppableThread, self).__init__(target=target)
def _stop(self):
"""
stop the thread, overwrite method, modified from threading.py
:return:
"""
lock = self._tstate_lock
self._is_stopped = True
self._tstate_lock = None
if not self.daemon:
with threading._shutdown_locks_lock:
threading._shutdown_locks.discard(lock)
此处注意,Python 版本至少为 3.7,如使用 Python 3.6 会报错 threading 模块没有 _shutdown_locks_lock 属性:
定义一个事件,并根据该事件写好线程的目标函数:
event_flag = threading.Event()
event_flag.set()
def thread_target():
"""
thread target function, inspect event flag
:return:
"""
global event_flag
while True:
if event_flag.is_set():
input("Press <enter> to pause:")
event_flag.clear()
event_flag.wait()
验证效果,主线程通过 for 循环计时,新线程监测键盘输入:
import time
if __name__ == '__main__':
thread = StoppableThread(target=thread_target)
thread.start()
tms = time.time()
for i in range(100):
print(i)
time.sleep(0.1)
if not event_flag.is_set():
s = input("Please input something:")
print("You input {}".format(s))
event_flag.set()
tme = time.time()
tm = round(tme - tms, 3)
print()
print("for loop running time: {} s".format(tm))
thread._stop()
|