大体的思路是：服务器启动时，查询数据库获取为推送的消息，存储到内存中，定义了一个全局变量MESSAGE_LIST，集合类型。以后的每次维护，包括(增、删、改)，都同时维护两份，及数据库和内存，保证两者的一致。当一条消息推送完成，即从MESSAGE_LIST中删除。
首先，添加spring的配置，在web-application-config.xml文件中添加
[html]
view plaincopy<flex:message-broker id="_messageBroker">
<flex:message-service
default-channels="my-streaming-amf,my-polling-amf"/>
</flex:message-broker>
<flex:message-destination id="tick-data-feed" channels="my-streaming-amf,my-polling-amf" allow-subtopics="true" subtopic-separator="."/>
没添加之前时：
[html]
view plaincopy<flex:message-broker />
主要是为了在后台获取message-broker，services-config.xml的配置还是一样，不变。
然后，后台java类：
[java]
view plaincopy/**
* 新增消息
* @param paramMessage
* @return
* @throws ServiceException
* @throws SQLException
* @author: IT氧吧
* @createDate: 2011-5-17
* @modifiedBy:IT氧吧
* @modifiedDate:2011-5-17
*/
public String savePushMessage(PushMessage paramMessage)
throws ServiceException, SQLException {
paramMessage.setStartDate(DateUtil.parseDate(paramMessage.getStartDateCN(), DateUtil.YMDHMS_PATTERN));
paramMessage.setEndDate(DateUtil.parseDate(paramMessage.getEndDateCN(), DateUtil.YMDHMS_PATTERN));
Date d = systemService.getDateFromDataBase();
if(paramMessage.getStartDate().compareTo(d) < 0 || paramMessage.getEndDate().compareTo(d) < 0){
throw new ServiceException("推送日期不能早于当前日期");
}
if(paramMessage.getStartDate().compareTo(paramMessage.getEndDate()) > 0){
throw new ServiceException("推送起始日期不能晚于推送结束日期");
}
/** 保存到数据库 */
Long messageId = systemService.savePushMessage(paramMessage);
/** 添加到集合中 */
GlobalNames.MESSAGE_LIST.add(paramMessage);
/** 集合重新排序 降序
Comparator<PushMessage> comparator = new Comparator<PushMessage>(){
@Override
public int compare(PushMessage o1, PushMessage o2) {
return o1.getStartDate().compareTo(o2.getStartDate());
}
};
Collections.sort(GlobalNames.MESSAGE_LIST, comparator); */
/**
* 如果GlobalNames.MESSAGE_LIST没有消息，此时新增消息的时候就得推送，否则只是把消息添加到GlobalNames.MESSAGE_LIST并排序
* */
if(GlobalNames.MESSAGE_LIST.size() == 1){
PushMessageUtil.startPush(systemService);
}
return messageId.toString();
}
下面看一下PushMessageUtil：
[java]
view plaincopypublic static void startPush(SystemService systemService){
if(GlobalNames.MESSAGE_LIST.size() > 0){
if(t == null){
t = new Timer();
try {
dbDate = systemService.getDateFromDataBase();
} catch (ServiceException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
/**
* 下面一段代码在一种情况下会报错，即系统刚启动还没有任何人登录，
* 此时系统flex框架部分还未初始化,获取到msgBroker为空
* */
msgBroker = MessageBroker.getMessageBroker("_messageBroker");
clientID = UUIDUtils.createUUID();
t.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
// TODO Auto-generated method stub
if(msgBroker == null){
msgBroker = MessageBroker.getMessageBroker("_messageBroker");
}
PushMessage message;
dbDate.setMinutes(dbDate.getMinutes()+1);
/** 每隔1分钟轮询一次 */
/** 注意下面的循环不能用foreach，否则报java.util.ConcurrentModificationException*/
for(PushMessage message : GlobalNames.MESSAGE_LIST){
for(int i=0;i<GlobalNames.MESSAGE_LIST.size();i++){
String dateStr = DateUtil.format(dbDate, DateUtil.YMDHMS_PATTERN);
message = GlobalNames.MESSAGE_LIST.get(i);
if((message.getStartDate().getMinutes()-dbDate.getMinutes()) % message.getInterval() == 0){
pMessage = message;
start();
}
//是否等于结束时间
if(message.getEndDateCN().substring(0, message.getEndDateCN().lastIndexOf(":")).equals(dateStr.substring(0, dateStr.lastIndexOf(":")))){
GlobalNames.MESSAGE_LIST.remove(message);
System.out.println("大小："+GlobalNames.MESSAGE_LIST.size());
stopMessage = message;
stop();
if(GlobalNames.MESSAGE_LIST.size() == 0){
t.cancel();
t = null;
break;
}
}
}
}
}, 1000*60, 1000*60);
}
}
/**
* 开始推送
* @author: IT氧吧
* @createDate: 2011-5-20
* @modifiedBy: IT氧吧
* @modifiedDate:2011-5-20
*/
private static void start() {
AsyncMessage msg = new AsyncMessage();
msg.setDestination("tick-data-feed");
msg.setHeader("DSSubtopic", "tick");
msg.setClientId(clientID);
msg.setMessageId(UUIDUtils.createUUID());
msg.setTimestamp(System.currentTimeMillis());
msg.setBody(pMessage);
if(msgBroker != null){
msgBroker.routeMessageToService(msg, null);
System.out.println("start!!");
}
}
/**
* 停止推送
* @author: IT氧吧
* @createDate: 2011-5-20
* @modifiedBy:IT氧吧
* @modifiedDate:2011-5-20
*/
private static void stop() {
System.out.println(stopMessage.getContent()+" 已经推送完成");
}
不解释，应该都能看懂，该有的注释也都有
前台还是一样，该添加的监听还是得添加，只是多了个，如果有多个消息同时推送时，已经推送过的消息将不会显示
[java]
view plaincopyprivate var msgList:ArrayCollection = new ArrayCollection();
private function messageHandler(event:MessageEvent):void
{
var pmsg:ServerPushMessageVO = event.message.body as ServerPushMessageVO;
/**
* 返回指示视图是否包含指定对象的信息。与 IViewCursor.findxxx 方法不同，
* 此搜索仅在找到完全与参数匹配的项目时才会成功。如果视图中包含已应用到该视图的滤镜，
* 则即使基本集合确实包含该项目，此方法也会返回 false。
* */
if(!msgList.contains(pmsg.id))
{
var showTipMessageG:ShowTipMessageG = new ShowTipMessageG();
msgList.addItem(pmsg.id);
showTipMessageG.serverPushMessageVO = pmsg;
PopUpEffect.show(showTipMessageG, this, true);
}
}
下面上几张效果图
新增消息页面
弹出提示页面
ok，消息推送的到此结束。。。
