内容简介：Californium开源框架之源码分析（三）
项目源码地址： https://github.com/eclipse/californium
摘要
物联网时代，为了在设备监控的数据发生变化时平台能第一时间获取到，频繁地定时地向设备获取其数据是不现实的。一是会消耗设备的电量、二是会浪费不必要的带宽。其解决方案是：平台作为一个客户端向设备（服务端）的数据发起一个订阅请求，在设备数据发生变化时主动推送给平台。交互图如下：
observe包
observe包就是上面讲到的订阅功能模块，实现客户端对服务端资源的订阅。其过程为：客户端发起一个订阅请求；服务端接收请求，找到对应的资源来处理请求，并保存该订阅关系；当服务端的资源发生变化时，服务端主动发送响应给客户端；客户端根据之前的订阅接收该响应。observe包图如下：
客户端相关
Observation类
该类表示一个观察，内部封装了Request请求和CorrelationContext上下文。
ObservationStore接口
该接口声明了对Observation对象进行存储，并对外提供了增删改查的公共方法。现在开发一个系统，为提高可靠性，通常都设计为多节点。框架提供该接口，就是希望开发者能够自己实现存储方式。例如，将Observation对象存储到数据库而不是内存，这样当系统中一个节点崩溃时，其他节点还能从数据库获取到Observation对象，即客户端还能处理之前订阅服务端后，服务端发来的响应消息。
当客户端发送请求消息并携带observe字段时，框架会保存该订阅请求。具体实现在 Matcher.sendRequest() 方法中，源码如下：
public void sendRequest(Exchange exchange,final Request request) {
// 忽略非关键代码
...
// 处理订阅请求
if (request.getOptions().hasObserve() && request.getOptions().getObserve() == 0 && ...) {
// 保存订阅请求到observationStore对象中
observationStore.add(new Observation(request, null));
// 监听请求，当请求取消、被拒绝、超时时，从observationStore对象中移除订阅请求
request.addMessageObserver(new MessageObserverAdapter() {
@Override
public void onCancel() {
observationStore.remove(request.getToken());
}
@Override
public void onReject() {
observationStore.remove(request.getToken());
}
@Override
public void onTimeout() {
observationStore.remove(request.getToken());
}
});
}
// 忽略非关键代码
...
}
当服务端接收订阅请求，会先发送一个订阅成功的响应。而后续的响应消息将由客户端的 Matcher.receiveResponse() 方法进行匹配检查，并通知NotificationListener。具体代码如下：
public Exchange receiveResponse(final Response response, final CorrelationContext responseContext) {
// 忽略非关键操作
...
// 根据响应消息中的token查找对应的订阅
final Observation obs = observationStore.get(response.getToken());
if (obs != null) {
// 获取之前发起的订阅请求消息
final Request request = obs.getRequest();
request.setDestination(response.getSource());
request.setDestinationPort(response.getSourcePort());
exchange = new Exchange(request, Origin.LOCAL, obs.getContext());
exchange.setRequest(request);
exchange.setObserver(exchangeObserver);
request.addMessageObserver(new MessageObserverAdapter() {
@Override
public void onResponse(Response response) {
// 通知订阅监听器
notificationListener.onNotification(request, response);
}
// 其他情况从observationStore对象中移除订阅请求
...
});
}
// 忽略非关键操作
...
}
InMemoryObservationStore类
该类实现了ObservationStore接口，从名字就可以看出它将Observation对象简单地存储在内存中，通过ConcurrentHashMap保存数据，其中key为KeyToken，value为Observation。
NotificationListener接口
客户端可以通过 Endpoint.addNotificationListener() 添加该监听器，当收到被订阅的服务端发来的响应时， onNotification() 方法将会被回调。
NotificationListener具有全局性。当添加了监听器后，所有资源的订阅响应都将回调，这是因为客户端订阅时是以服务端的资源为单位的，而监听器是在客户端的Endpoint里添加的，关系如下图：
当然框架也提供了一个一对一关系的回调，通过 CoapClient.observe(Request request, CoapHandler handler) 方法实现，这里就不展开了。
服务端相关
ObserveRelation类
由于客户端订阅服务端资源，所以服务端需要存储客户端的订阅信息。该类表示客户端的Endpoint与服务端的Resource对应关系。与下面我们要讲的 ObservingEndpoint类 关系紧密。
ObservingEndpoint类
该类表示客户端发起订阅的Endpoint，它包含着一个客户端与服务端所有资源建立的订阅关系，所以与 ObserveRelation类 是一个一对多的关系。为了形象化，画了下面这个图供大家参考：
当一个CON类型的订阅响应发送给客户端超时之后，服务端可以认为客户端已不可达，并解除所有已经建立的订阅关系。
以上所述就是小编给大家介绍的《Californium开源框架之源码分析（三）》，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持！
猜你喜欢:
- 优秀开源库SDWebImage源码浅析
- 优秀开源库SDWebImage源码浅析
- Californium开源框架之源码分析（四）
- Android开源框架源码分析：Okhttp
- Android开源框架源码分析：Okhttp
- nuxt全栈实践 开源后台源码
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
