flex_blazeds推送机制

flex blazeds推送机制 前言： 对于数据实时性页面呈现非常强的业务，通过前台定时刷新这样的效率明显不行。下面通过 flex技术来实现这个功能。 思路： 以告警数据为例，告警数据通过mq、socket等方式进入系统，在 java服务器通过监听来 监听消息队列数据，接收后推送到 flex各个客户端。blazeDS采用 amf 协议 离婚协议模板下载合伙人协议 下载渠道分销协议免费下载敬业协议下载授课协议下载 。该协议可以 传输 Object, Array, Date, XML。由于 AMF采用二进制编码，这种方式可以高度压缩数据， 因此非常适合用来传递大量的资料。数据量越大，Flash Remoting的传输效能就越高，远 远超过Web Service。至于 XML, LoadVars和 loadVariables() ，它们使用纯文本的传输方 式，效能就更不能与 Flash Remoting相提并论了 本例mq使用 activeMq测试，推动到 flex的方式采用adobe提供的 java flex通信包 blazeDS 中提供的推送功能， 具体步骤： 一． 创建一个 java flex使用 blazeDS通信的工程，具体步骤在此不详细说明搭建的步骤。 搭建成功后导入 activemq-all-5.1.0.jar，为后面接收 activemq消息使用。 二． 配置： 在 remoting-config.xml加入推送服务。 com.east.flex.serverpush.chart.ServerPushServer 1.服务发布 Java代码：ServerPushServer.Java Java代码 1. package com.east.flex.serverpush.chart; 2. 3. /** 4. * 服务发布类 5. * 6. * @author East(姜可亮) 7. * 推送方式参考于网络，感谢网络。自己添加了 mq驱动。 8. */ 9. public class ServerPushServer { 10. //ServerPushThread serverpushthread = new ServerPushThread(); 11. MessageBrokFactory messageBrokFactory; 12. public void controlThread(String str){ 13. if(str.equals("start")){ 14. try{ 15. messageBrokFactory=new MessageBrokFactory(); 16. messageBrokFactory.start(); 17. }catch (Exception e){ 18. e.printStackTrace(); 19. } 20. 21. //serverpushthread.start(); 22. }else{ 23. messageBrokFactory.stop(); 24. //serverpushthread.stop(); 25. } 26. } 27. } 2.消息处理事务: messageBrokFactory 具有功能：启动mq,并监听mq接收消息,消息推送到 flex.三大功能。 Java代码 1. package com.east.flex.serverpush.chart; 2. 3. import javax.jms.JMSException; 4. 5. import javax.jms.Message; 6. import javax.jms.MessageListener; 7. import javax.jms.TextMessage; 8. import javax.naming.NamingException; 9. 10. import org.apache.log4j.Logger; 11. 12. import com.zznode.gixpert.provider.activemq.service.impl.ActiveMQAdapter; 13. 14. import flex.messaging.MessageBroker; 15. import flex.messaging.messages.AsyncMessage; 16. import flex.messaging.util.UUIDUtils; 17. 18. 19. /** 20. * 消息处理事务 21. * 功能：启动 mq,并监听 mq接收消息,消息推送到 flex.三大功能。 22. * @author 姜可亮 23. * 2011-3-23 24. * 25. * **/ 26. public class MessageBrokFactory { 27. /** 28. * flex消息组件 29. */ 30. private MessageBroker msgBroker; 31. /** 32. * flex uuid 33. * */ 34. private String clientId ; 35. 36. /********************以下为 mq初始化变量****************************/ 37. /**端口*/ 38. private static int port=61616; 39. /**mq服务器主机 ip/主机名*/ 40. private static String hostname = "localhost"; 41. /**activemq用户名*/ 42. private static String user = ""; 43. /**activeMq密码*/ 44. private static String password= ""; 45. /** 46. * 队列名称 47. */ 48. private String queuename="sendToFlex"; 49. /** 50. * activemq服务器 51. * */ 52. ActiveMQAdapter amqAdapter; 53. /** 54. * 采用的接受方式 默认是 Queue 55. * 一个消息向多个客户端发送使用 topic 56. * 一个消息只发送到一个客户端使用 Queue 57. * */ 58. private static boolean pattern; 59. /** 60. * 是否采用消息持久化 61. * 暂时不用 62. * */ 63. private static String durable="是"; 64. /** 65. * 睡眠时间 66. * 暂时不用 67. * */ 68. private static String sleepTime="否"; 69. /** 70. * 超时时间 71. * 暂时不用 72. */ 73. private static String receiveTimeOut="否"; 74. /***********************初始化变量结束************************************ / 75. 76. private Logger logger = Logger.getLogger(ActiveMQAdapter.class); 77. 78. public boolean sedMessage(String sub,Object message) { 79. try{ 80. AsyncMessage msg = new AsyncMessage(); 81. msg.setDestination("serverpush"); 82. msg.setClientId(clientId); 83. msg.setMessageId(UUIDUtils.createUUID()); 84. msg.setTimestamp(System.currentTimeMillis()); 85. /* 设置消息信息，发布到客户端去 */ 86. msg.setBody(message); 87. msgBroker.routeMessageToService(msg, null); 88. return true; 89. }catch(Exception e){ 90. e.printStackTrace(); 91. return true; 92. } 93. 94. } 95. /* 96. * 启动 activeMq采集 97. * 98. * */ 99. public void start(){ 100. try { 101. //初始化 flex消息组建 102. msgBroker = MessageBroker.getMessageBroker(null); 103. clientId = UUIDUtils.createUUID(); 104. //启动 mq消息接收 105. amqAdapter = new ActiveMQAdapter(); 106. //版本 1.0只需要对 hostname port 起作用 有更高需求再另添加 107. amqAdapter.setUrl("failover:(tcp://"+hostname+":"+port+"?wireFo rmat.maxInactivityDuration=0)"); 108. 109. amqAdapter.addMessageListener(queuename, pattern, new MyTestMes sageListener()); 110. } catch (JMSException e) { 111. String msg = "Cannot start AlarmCollectQueue."; 112. logger.error(msg, e); 113. System.out.println("set config ex"+e); 114. e.printStackTrace(); 115. } catch (NamingException e) { 116. String msg = "Naming context error."; 117. logger.error(msg, e); 118. System.out.println("set config ex"+e); 119. e.printStackTrace(); 120. } 121. } 122. /* 123. * 停止 activeMq采集 124. * 125. * */ 126. public void stop() { 127. amqAdapter.close(); 128. } 129. /** 130. * 消息接收处理工厂 131. * */ 132. class MyTestMessageListener implements MessageListener { 133. 134. public void onMessage(Message msg) { 135. try{ 136. // 137. String message=((TextMessage)msg).getText(); 138. logger.info("recieve a message from: "+msg.getJMSDestinatio n()+" " + 139. "message text: "+message); 140. sedMessage("test",message); 141. }catch(Exception e){ 142. e.printStackTrace(); 143. System.out.println("set config ex"+e); 144. } 145. } 146. } 147. 148. } 2.ActiveMq接收消息的代码见附件工程。在此不列出。 Flex端：本例以一个 textarea文本来显示数据。告警数据具体可以使用 datagrid 展现代码 getMessageClient.mxml Flex代码 1. 使用操作： 在第一次打开的页面点击 startserver.这个共是启动监听，注意只需要在第一次打开点击了 就行，其他打开的页面不需要。再次点击 然后点击 autorefesh，用来启动自动显示数据，注意 autorefesh这不是自动刷新请求数据。 是推动上来的数据。 其他打开的页面都只要点击 autorefesh即可。 启动一个 acitmvemq在提供的页面发送数据，可以一次发送几千条数据。同事打开几个页 面来测试下下过。 项目最近开始运行发现客户端 web非常耗费 cpu,非常头疼，当然这与我的数据量肯定是有 关系的，我后台推送到前台的数据大约每秒 100条。消息量少时 streaming没有问 题 快递公司问题件快递公司问题件货款处理关于圆的周长面积重点题型关于解方程组的题及答案关于南海问题 。 后来经过了多种情况的测试，最后发现把 streaming设置为 polling方式轮询推送占用的 cpu 资源一下子降了下来。后来吧轮询时间这只为 0.3s效果比 streaming还好。streaming方式 是，消息上来的太快， 表 关于同志近三年现实表现材料材料类招标技术评分表图表与交易pdf视力表打印pdf用图表说话 pdf 格显示时非常闪眼睛，用户没法操作，相反设为 polling后这种现 象降低了。 总结 初级经济法重点总结下载党员个人总结TXt高中句型全总结.doc高中句型全总结.doc理论力学知识点总结pdf steaming耗费 cpu大的原因是，flex执行死循环时比 java执行死循环耗资源打多了。 也就是对于一直占用 cpu不释放。steaming方式消息量大了也跟执行死循环一样，一直在 执行不释放 cpu. 下面是改造后的配置 Java代码 1. 2. 3. 4. true 5. 0.2 6. 7. 8.