Floodlight中 处理packetin消息的顺序(2)
消息 处理 顺序
2023-09-11 14:20:43 时间
前面通过阅读代码知道了怎样推断各个模块处理某个消息的先后顺序。那么内部是怎样实现的呢?
每当一个模块表示对一个消息感兴趣的时候,就会调用IFloodlightProviderService(详细有Controller类实现)的addOFMessageListener方法进行注冊订阅,核心工作是由 ListenerDispatcher类来完毕:1)每次添加一个观察者的时候都会推断其是否是终结点(也就是不被其它的listener所依赖),由于终于确定这些观察者顺序的时候就是由这些终结点開始往前进行DFS遍历而得到。2)比方说ForwardingBase和Distributing
(我们自己加的。没有约束其顺序)。当它们注冊packetin消息的时候。会增加到终结点集合terminals中。所以从它们開始深度遍历的时候得到的有序集合ordering=linkdiscovery,topology,devicemanager, forwarding, distributing(这里进行了两次DFS traverse)。接下来看代码:
-------------Controller中实现IFloodlightProviderService的方法
@Override
public synchronized void addOFMessageListener(OFType type,
IOFMessageListener listener)
{
//先推断与type相应的 ListenerDispatcher对象是否存在
ListenerDispatcher< OFType, IOFMessageListener>
ldd =
messageListeners.get(type);
if (ldd
== null ) {
ldd = new ListenerDispatcher< OFType, IOFMessageListener>();
messageListeners.put(type, ldd);
}
//注冊监听type这个消息。
ldd.addListener(type, listener);
}
--------------ListenerDispatcher实现(维护这些观察者,有依赖关系)
public class ListenerDispatcher <U,
T extends IListener<U>>
{
protected static Logger logger = LoggerFactory.getLogger(ListenerDispatcher. class );
List<T> listeners = null;
//每一个OF msg都有唯一的ListenerDispatcher对象。观察者存在listeners链表中
//从listener这个观察者開始,根据有没有监听者在他之前,进行深度优先遍历
//终于有序序列存在ordering中。visited用于存已经訪问过的terminal
listener。
private void visit(List<T>
newlisteners, U type, HashSet<T> visited,
List<T> ordering,
T listener) {
if (!visited.contains(listener))
{
visited.add(listener);
for (T
i : newlisteners) {
if (ispre(type,
i, listener)) {
visit(newlisteners, type, visited, ordering, i);
}
}
ordering.add(listener);
}
}
//推断观察者l1
是否在 l2 之前(每一个观察者实现了IListener接口)
private boolean ispre(U
type, T l1, T l2) {
return (l2.isCallbackOrderingPrereq(type,
l1.getName()) ||
l1.isCallbackOrderingPostreq(type, l2.getName()));
}
//订阅type消息。
public void addListener(U
type, T listener) {
List<T> newlisteners = new ArrayList<T>();
if (listeners != null)
newlisteners.addAll( listeners );
newlisteners.add(listener);
//
Find nodes without outgoing edges
List<T> terminals = new ArrayList<T>();
for (T
i : newlisteners) {
boolean isterm
= true;
for (T
j : newlisteners) {
if (ispre(type,
i, j)) {
isterm = false ;
break ;
}
}
if (isterm)
{
terminals.add(i); //维护终节点集合
}
}
if (terminals.size()
== 0) {
logger .error("No
listener dependency solution: " +
"No
listeners without incoming dependencies");
listeners =
newlisteners;
return ;
}
//接下来得到有序的listeners;
//
visit depth-first traversing in the opposite order from
//
the dependencies. Note we will not generally detect cycles
HashSet<T> visited = new HashSet<T>();
List<T> ordering = new ArrayList <T>();
for (T
term : terminals) {
visit(newlisteners, type, visited, ordering, term);
}
listeners =
ordering;
}
//观察者退出。为何不直接remove??
public void removeListener(T
listener) {
if (listeners != null)
{
List<T> newlisteners = new ArrayList<T>();
newlisteners.addAll( listeners );
newlisteners.remove(listener);
listeners =
newlisteners;
}
}
//清除全部listeners;
public void clearListeners()
{
listeners = new ArrayList<T>();
}
//
public List<T>
getOrderedListeners() {
return listeners ;
}
}
数据结构关联图:
相关文章
- 【C#】对异步请求处理程序IHttpAsyncHandler的理解和分享一个易用性封装 【手记】走近科学之为什么明明实现了IEnumerable<T>的类型却不能调用LINQ扩展方法 【手记】手机网页弹出层后屏蔽底层的滑动响应 【手记】ASP.NET提示“未能创建类型”处理 【Web】一个非常简单的移动web消息框 【手记】解决EXCEL跑SQL遇“查询无法运行或数据库表无法打开...”
- 微服务轮子项目(33) -RocketMQ特点、安装部署、异常处理、事务消息原理
- 微信小程序 - 唤起订阅消息(模板消息)介绍及处理相关操作
- 用Python发送通知到企业微信,实现消息推送
- 一分钟读懂MySQL分布式消息的处理
- CSDN日报191120:《吊打面试官》系列-消息队列基础;Vue + Spring Boot 项目实战
- Windows 消息ID及定义大全
- C# winform 禁用最小化和还原按钮(消息循环截获处理)
- MFC测试Windows消息处理机制、阻塞试验、SetTimer、MessageBox、消息响应处理函数
- SendMessage消息发送原理与死锁处理
- MFC自定义消息的实现方法
- QT中处理不同Windows(窗体中的)消息
- ROS节点发布消息控制RVIZ中的MAKER
- Spring Cloud Stream同一通道根据消息内容分发不同的消费逻辑
- 《C++面向对象高效编程(第2版)》——1.4 理解消息、方法和实例变量
- Android应用程序键盘(Keyboard)消息处理机制分析
- android的消息处理机制(图+源码分析)——Looper,Handler,Message
- IIS 分析器错误消息: 未能加载类型“_Default”
- runtime - 消息发送(objc_msgSend)
- android学习-异步消息处理机制
- 浅析微信公众号订阅消息开发流程
- SpringBoot集成ActiveMq消息队列实现即时和延迟处理
- 阿里巴巴消息系统架构与变迁--转载
- FreeRTOS消息队列、信号量、事件组、任务通知之间的区别
- VCL消息处理机制