Python实现Event回调机制的方法
时间：2020-11-11 11:59:49 | 栏目：Python代码 | 点击：次
0.背景
在游戏的UI中，往往会出现这样的情况：
在某个战斗副本中获得了某个道具A，那么当进入主界面的时候，你会看到你的背包UI上有个小红点（意思是有新道具），点击进入背包后，发现新增了道具A，显示个数为1，并且在下个界面中有个使用的按钮由灰色不可使用变成橙色的可使用状态
图1. 事件触发说明图
其中这里是由道具获得这个事件，触发了上述的三个行为。如果使用显示调用行为，会使得代码难扩展，易出错，逻辑混乱等问题，如果使用Event回调机制，就会变得十分方便。
其实Event回调机制就是观察者模式，如下图：
图2. 观察者模式
在C#中存在（delegate & event）的语义来实现Event回调机制：具体使用如下：
public delegate void NewToolGotEvent();
public class ToolBag
{
event NewToolGotEvent newToolGotHandler;
void Start()
{
newToolGotHandler += renderRedPoint;
newToolGotHandler += renderNewTool;
newToolGotHandler += renderAvaliableUseBtn;
}
void renderRedPoint()
{
//TODO
}
void renderNewTool()
{
//TODO
}
void renderAvaliableUseBtn()
{
//TODO
}
void EventHappened()
{
newToolGotHandler(); // usage, fill args if necessary
}
}
如果在Python，可以在注册事件的回调时，带入一个参数callback，在注册函数实体内，存在一个list将callback添加进去，形如：
def register_callback(self, cb): self.callbacks.append(cb)
但是这样是一个最为普遍的做法，既然是Python，这里我们有更Pythonic的做法，而且相比于上述的观察者模式，它的做法更加简洁，使用更加方便，接下来我们来解析一下Python实现Event callback的步骤。
1. UML类图
上述案例中，是针对游戏客户端UI的案例。所以我们呈现出的UML图也是与UI相关。如图3所示，它显示了Python中实现Event回调的机制。
图3. UML关系图
如上图所示，此机制主要由三个类及他们的实例（instance）组成：UIBase, UIScene, UIDataEvent。
1 . UIBase: 所有UIScene的基类，其实例有scene_id变量，包含两个必要的方法， __init__ 是初始化方法，init_data_listeners方法是将实例中的某些方法, 例如ui_updata_func中包含的UIDataEvent实例（所有的UIDataEvent实例都是单例）遍历，并把ui_update_func注册在每一个UIDataEvent实例中。
2 . UIScene: 场景类，管理某个场景的UI渲染。在其实例中，存在某些方法，例如ui_update_func需要在某些UIDataEvent实例触发时候，也被同时触发调用。ui_update_func在Python中一个bound method object， 它会拥有一个特殊的属性events，即所有需要触发此方法的UIDataEvent实例集合。这个通过装饰器（decorator）来实现,即图中的：
“ui_update_func” is a Python object which add a amount of UIDataEvent instances by Python decorator named “data_listener”
3 . UIDataEvent: 事件类，该类有个类变量_events, 记录了所有的UIDataEvent实例，每一个UIDataEvent实例都是单例，而且都有一个名字，和一个回调方法集合_callbacks, 里面的每一个方法都是在本事件触发后需要回调的方法。实例还有个__iadd__方法，将需要回调的函数cb注册进去。__call__事件触发是实际触发的函数。
2. 代码
上一步讲述了三个类之间的联系与各自的作用，此步展示代码实现相关功能。
a) UIBase.py
首先列出来的是UIBase的类，除了上述的__init__与init_data_listeners方法，还多了destroy方法
# -*- coding: utf-8 -*-
from UIDataNotifier import UIDataEvent
import inspect
class UIBase(object):
def __init__(self, in_scene_id):
self.id = in_scene_id
self.init_data_listeners()
def init_data_listeners(self):
"""为所有标有@data_listener的成员函数注册事件监听器"""
for listener_name, listener in inspect.getmembers(self, lambda f: hasattr(f, 'events')):
for event in listener.events:
event += listener
def destroy(self):
print '%s.destroy' % self.__class__.__name__
UIDataEvent.clear()
init_data_listener比较难理解，我们看一下built-in的inspect.getmembers的源码：
def getmembers(object, predicate=None): """Return all members of an object as (name, value) pairs sorted by name. Optionally, only return members that satisfy a given predicate.""" results = [] for key in dir(object): try: value = getattr(object, key) except AttributeError: continue if not predicate or predicate(value): results.append((key, value)) results.sort() return results
其实源码的意思就是，在dir(object)的value中找，找到能够满足predicate(value) == True的value,然后将(key, value)收集，进行排序后返回。
放在代码的意思是：
for listener_name, listener in inspect.getmembers(self, lambda f: hasattr(f, 'events')):
for event in listener.events:
event += listener
在dir(scene)中找，找到value中存在名叫events的属性, 返回得到是一个list，每个list的元素是一个二元tuple: (key, value)，其中key，即listener_name是dir(scene)的属性名，而value， 即listener就是属性对象，这里其实就是包含事件的函数对象，然后遍历listener中的每一个UIDataEvent实例，并将listener注册到event中（+= ==> __iadd__ ）
b) UIScene.py
UIScene的代码如下：
# -*- coding: utf-8 -*- from UIDataNotifier import * from UIBase import UIBase class UIScene(UIBase): def __init__(self, in_scene_id): super(UIScene, self).__init__(in_scene_id) @data_listener(OnItemAdded) def ui_render_red_point(self, item): print 'ui_render_red_point' @data_listener(OnItemAdded) def ui_render_new_tool(self, item): print 'ui_render_new_tool: ' + item @data_listener(OnItemAdded) def ui_render_avaliable_use_btn(self, item): print 'ui_render_avaliable_use_btn' bag_ui_scene = UIScene(123)
在UIScene中只是要填写对于OnItemAdded这个事件触发之后，需要回调的函数，上述代码中写了三个函数。注意需要在函数上加上装饰器@data_listener(OnItemAdded)，这样此函数就会添加一个特殊的属性events，具体装饰器的代码见UIDataNotifier.py。
最后新建一个bag_ui_scene的scene。
c) UIDataNotifier.py
UIDataNotifier.py代码如下：
# -*- coding: utf-8 -*-
import sys
def data_listener(*events):
def wrapped_f(f):
f.events = events
return f
return wrapped_f
class UIDataEvent(object):
_events = []
def __init__(self, name):
self._name = name
self._callbacks = []
UIDataEvent._events.append(self)
def __iadd__(self, cb):
self._callbacks.append(cb)
return self
def __call__(self, *args, **kwargs):
for cb in self._callbacks:
try:
cb(*args, **kwargs)
except:
ex = sys.exc_info()
print "UIDataNotifier cb error, function:", cb.__name__, ex
def __repr__(self):
return 'UIDataEvent %s' % self._name
@classmethod
def clear(cls):
"""清空所有事件上的所有监听器，在销毁一个界面的时候调用"""
for event in cls._events:
event._cb = []
OnItemAdded = UIDataEvent('OnItemAdded')
data_listener装饰器其实就是声明一个特殊的events属性，并将所有在UIScene中填写的UIDataEvent实例元组集合赋值给它。
__iadd__是将参数cb添加到实例的变量中_callbacks中，此方法在UIBase的init_data_listeners中使用。
__call__是当UIDataEvent实例自调用时，例如OnItemAdded(item)，实际调用的函数，在函数体里，会回调_callbacks中的每个方法，这也就是Event回调机制的核心部分，相当于观察者模式的notify方法
最后新建一个OnItemAdded事件。
c) client.py
创建上述几个类之后，使用Event回调就非常简单了，代码如下：
# -*- coding: utf-8 -*-
from UIScene import UIScene
from UIDataNotifier import *
OnItemAdded('liu_xin_biao') #新道具流星镖获得事件发生了
输出：
ui_render_avaliable_use_btn ui_render_new_tool: liu_xin_biao ui_render_red_point
3.使用方法
1. 在本模块内增加一个事件定义，并在注释中写明事件的参数及意义。
如果要监听一个事件，请仔细阅读相关注释。
2. 在ui类最顶端import需要的事件及data_listener。
3. 在需要响应该事件的方法（监听器方法）前增加装饰器@data_listener，参数内列出要监听的所有事件。
如：
@data_listener(OnEventA, OnEventB) def my_listener_method(arg1): ...
注意保持监听器方法的参数个数及意义与事件触发的地方一致。
4. 在逻辑代码中适当的位置对事件进行触发。如OnEventA(arg1, ...)
注意：并不是所有与UI的交互都必须使用事件，事件机制是为了方便多对多的交互。比如背包物品改变事件，有多个UI都会监听背包物品的变化，而有多种逻辑都会导致背包物品变化，这时使用事件就比较方便。
4. 总结
本文主要讲述了如何使用Python实现Event回调机制，上述的示例代码参考我的[github-EventCallBack] (https://github.com/csdz/SnapToSnap/tree/master/EventCallBack)。