![结合使用 call_soon_threadsafe 和 GLib.idle_add 是从 asyncio 安全调用 GTK 代码的正确方法吗,还是有点过度?](https://linux22.com/image/1186129/%E7%BB%93%E5%90%88%E4%BD%BF%E7%94%A8%20call_soon_threadsafe%20%E5%92%8C%20GLib.idle_add%20%E6%98%AF%E4%BB%8E%20asyncio%20%E5%AE%89%E5%85%A8%E8%B0%83%E7%94%A8%20GTK%20%E4%BB%A3%E7%A0%81%E7%9A%84%E6%AD%A3%E7%A1%AE%E6%96%B9%E6%B3%95%E5%90%97%EF%BC%8C%E8%BF%98%E6%98%AF%E6%9C%89%E7%82%B9%E8%BF%87%E5%BA%A6%EF%BC%9F.png)
我有一个在 asyncio 事件循环中运行的应用程序,我想添加一个与我的 Gnome Shell 交互的功能,特别是发送通知。由于 GTK 有自己的(我理解是不兼容的)事件循环,我决定将其放入Gio.Application
后台线程。我的问题是关于从我的 asyncio 事件循环运行“GTK 代码”的适当方法,例如创建通知并从我的应用程序中触发它。
以下面的代码为例,我使用以下代码行来调度“添加通知”方法:
asyncio.get_event_loop().call_soon_threadsafe(GLib.idle_add,
app.add_notification, "You have a new email")
这“有效”,但我想知道call_soon_threadsafe
和是否idle_add
都是多余的。如果我简单地调用它也可以有效
app.add_notification("You have a new email")
从我的协程,然而PyGObject 文档表明这是线程不安全的。推荐的安全方法是什么?下面是从 asyncio 协程启动通知的最小概念验证。
import asyncio
import threading
import gi
gi.require_version('Gio', '2.0')
from gi.repository import Gio, GLib
class ThreadsafeEvent(asyncio.Event):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if self._loop is None:
self._loop = asyncio.get_event_loop()
def set(self):
self._loop.call_soon_threadsafe(super().set)
def clear(self):
self._loop.call_soon_threadsafe(super().clear)
class Application(Gio.Application):
APP_ID = "org.something"
def __init__(self, activated_event: threading.Event):
flags = Gio.ApplicationFlags.FLAGS_NONE
super().__init__(application_id=Application.APP_ID, flags=flags)
self.activated_event = activated_event
self.activated = False
def do_activate(self):
if not self.activated:
self.hold()
self.activated = True
self.activated_event.set()
def add_notification(self, message):
note = Gio.Notification.new(message)
self.send_notification(None, note)
async def main():
event = ThreadsafeEvent()
app = Application(event)
def start(app: Application):
app.register()
app.run()
thread = threading.Thread(target=start, args=(app,))
thread.setDaemon(True)
thread.start()
try:
# When the event is set, we know the Application
# is activated and safe to send notifications through
await asyncio.wait_for(event.wait(), 10)
except asyncio.TimeoutError:
print('Timed out before App received its activation.')
asyncio.get_event_loop().call_soon_threadsafe(GLib.idle_add,
app.add_notification, "You have a new email")
if __name__ == '__main__':
loop = asyncio.get_event_loop()
asyncio.ensure_future(main(), loop=loop)
loop.run_forever()