我正在尝试返回Daemon线程的结果。我使用evdev来监控触摸输入事件,如果有超过0个触摸事件,那么我想返回True。我试图将列表作为arg传递给线程的目标函数,在这个版本中,我将添加到列表中。然而,我在第一次触摸时一直得到错误RuntimeError: There is no current event loop in thread
。
import time
def check_for_input(result):
from evdev import InputDevice, codes
dev = InputDevice('/dev/input/event0')
for event in dev.read_loop():
if event.type == ecodes.EV_KEY:
result.append([True])
from threading import Thread
results = [False]
input_thread = Thread(target=check_for_input, args=(results), daemon=True)
input_thread.start()
# do some other stuff
time.sleep(5)
input_thread.join()
print(results)
我该如何解决这个问题?一旦我解决了这个错误,它会起作用吗?另外一个问题是,这是最好/最简单的方法吗?
编辑:我已经设法通过在线程中创建一个循环来消除运行时错误。此外,使用Bhargav的建议,我可以得到作为全局变量的结果。所以我的代码目前是这样的:
import time
import asyncio
def check_for_input(stop_event):
print('started')
global results
from evdev import InputDevice, ecodes
dev = InputDevice('/dev/input/event0')
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
for event in dev.read_loop():
if event.type == ecodes.EV_KEY:
results = True
break
if stop_event.is_set():
break
print('EXITING')
from threading import Thread, Event
stop_event = Event()
stop_event.clear()
results = False
input_thread = Thread(target=check_for_input, args=(stop_event,), daemon=True)
print('about to start')
input_thread.start()
# do some other stuff
time.sleep(5)
stop_event.set()
print(results)
然而,根据我的补充问题,这是正确的做法吗?我相信人们会说这里要避免全局变量。此外,如何在print(results)
之前杀死线程?上面的stop_event.set()
似乎不起的作用
我认为这对你有用:
import time
def check_for_input():
global results
from evdev import InputDevice, codes
dev = InputDevice('/dev/input/event0')
for event in dev.read_loop():
if event.type == ecodes.EV_KEY:
results = True
from threading import Thread
results = False
input_thread = Thread(target=check_for_input, daemon=True)
input_thread.start()
# do some other stuff
time.sleep(5)
input_thread.join()
print(results)