基于Redis的Notification设计

需求：

这个功能开发的直接需求是为了提醒操作员即使处理库位补货， 在用户操作分拣波次操作以后， 会出现有库位库存为负数， 这种情况下需要有消息通知机制通知相关人员对相应库位进行补货处理；

1. 通知实体: 被通知的用户实体可能是具体到某些登录用户, 也可能是某些Role 下的用户
2. 通知状态: 消息窗口会采用浮动窗口告知用户有N条记录未读， 以及未读消息摘要； 消息在被用户阅读之前是未读状态， 读取以后会改变状态为已读状态； 消息窗口不会通知用户已读消息。
3. 消息定制： 应该提供能力供实施部门方便的定制一些业务规则， 来添加简单的新的消息类型和新的消息产生机制， 比如实施部门可以配置定时任务扫描DB中的业务表， 根据规则发现异常数据， 并把相关信息以消息形式发送给WMS 客户端；

设计：

总体设计

客户端如何获取消息：

  主流的方式无法长连接推送和客户端轮询的拉取的方式， 为了简化Server 和客户端的开发复杂度， 我们选择了客户端轮询的方式， 这样可以直接沿用我们WMS 中已有的WebService 接口的方式来暴露消息；

存储：

 消息体：

   考虑到通知消息属于非核心业务， 并且对轮询的方式会产生大量的请求， 所以我们不打算使用DB 做消息的存储， 以减少大量低优先级的db 访问来拖累DB 性能和稳定性；所以我们采用redis 来存储每个订阅者订阅的消息；

 消息订阅定义：

  要实现消息订阅的可配置， 需要对消息类型， 仓库， 订阅者进行关联， 这些信息存储在DB中， 在系统启动的时候会把这些信息Load 到redis 中， 后续的界面操作针对这个数据的增删改会同步修改redis 中的缓存数据；

数据格式：

1. 客户端访问App Server的接口是以xml 格式的方式进行对接；

2. Server 内部存储在Redis 中的数据以protobuf 的格式进行序列化和反序列化；具体在实现上使用Protostuff 生成protobuf runntime schema 对java 对象notification 进行序列化反序列化

数据结构

1. DB

 消息订阅定义：

  已有表添加数据:

  GV_SYS_CODECLASS 中添加 code = 'SUBSCRIBE_TYP',  CNCATEGROYNAME = '订阅类别的记录'

  GV_SYS_CODEINFO 中添加 code = 'REP_NOTICE', CNCATEGROYNAME  =  '补货通知', codeclass_id = #1 中的ID  的记录

  创建新表 GV_SUBSCRIBER

CREATE TABLE GV_SUBSCRIBER
 ( "ID" NUMBER(19,0) NOT NULL ENABLE,
 //其他省略
 "WHID" NUMBER(19,0),
 "SUBSCRIBERID" VARCHAR2(255 CHAR),
 "SUBSCRIBERTYPE" VARCHAR2(255 CHAR),
 "MESSAGETYPE_ID" NUMBER(19,0),
  PRIMARY KEY ("ID")

 )

2. Redis

    Redis 中存储的

 WMS_SUBSCRIBERS -->Set of subscriber

 WMS_SUBSCRIBER_TYPE _$messageTypeId_$whID --> Set of subscriber

 WMS_SUBSCRIBERED_MESSAGES_$subscriberId_$whID --> Ordered set of notification

 以 “WMS_SUBSCRIBERS” 为key 存储 Subscriber 集合， 这个集合的目的是为了能够取到当前系统中所有subscriber， 然后可以遍历 WMS_SUBSCRIBERED_MESSAGES_$subscriberId_$whID 键值列表， 删除每个键值所对应的有序集合中的过期通知； WMS_SUBSCRIBER_TYPE _$messageTypeId_$whID 键值对存储的是从表GV_SUBSCRIBER 中加载的订阅者集合， 每种类型的通知在每个仓库中的订阅者（User 或Role）

其中WMS_SUBSCRIBERED_MESSAGES_$subscriberId_$whID 往Ordered Set 中新增element 的时候以该notification 当前产生的时间戳为排序字段；

接口

1.通知消息产生：

通知消息产生的业务方需要知道当前消息的通知类别ID 和需要发送的逻辑仓库ID

 测试用例中的模拟代码如下：

 Notification notice = new Notification();
 notice.setId(UUID.randomUUID().toString());
 notice.setTitle("消息标题" + dateformat.format(new Date()));
 notice.setBody("消息体");
 notice.setCreateTime(System.currentTimeMillis());  //必须是当前时间戳
 notice.setExpireTime(60*30);
 notice.setWhId(119240L);
 notice.setMessageTypeId(124719L);

nManager.addNotice(notice);

NotificationManager 会从redis 缓存中获取 WMS_SUBSCRIBER_TYPE _$messageTypeId_$whID所对应的所有Subscriber，然后会针对每一个Subscriber 调用redis 接口

向有序集合   WMS_SUBSCRIBERED_MESSAGES_$subscriberId_$whID --> Ordered set of notification 中添加Notfication

2. 前端消息获取：

接口: PackNotificationVo listNotifications(
 ClientProperty clientProperty, Long whId, List<Long> subscriberIdList, Long lastFetchTimeStamp)

 客户端程序需要维护 whid， lastFetchTimeStamp 信息在本地， 每次请求把这个时间戳信息发送给服务器端，

客户端行为的伪代码就是

lastFetchTimeStamp = get（$whid）;

$newTimeStamp = 新的时间戳;


if($lastFetchTimeStamp& ==null ) $lastFetchTimeStamp = 默认的当前时间- 3天

call Server

展示新消息

set ($whid, 返回Notification 对象列表中创建时间最大的时间戳 )

服务器端的行为会把Notification在给定时间戳之后的 Ordered set 中的元素返回给客户端

 

过期消息删除

过期消息删除使用定时任务， 根据给定TTL， 即时出当前时间减去TTL 的时间得到超时时间点， 在这个时间点之后的Notification 都应该被删除；

 

maxScore = nowTimeStamp - TTL ;

shardedJedis.zremrangeByScore(WMS_SUBSCRIBERED_MESSAGES_$subscriberId_$whID, 0D, maxScore);

 

 

 

 

评论

1 楼 一江春水邀明月 2013-01-17  

tumblr也是选择类似的设计做消息通知 http://www.w3ccollege.org/redis/staircar-tumblr-of-redis-cluster-control-layer.html

不过他的方案是按Range 删除， 我的方案是按照超时按Score 删除。