Flex Blazeds推送机制

发布时间：2020-12-15 04:44:49 所属栏目：百科来源：网络整理

导读：前言：对于数据实时性页面呈现非常强的业务，通过前台定时刷新这样的效率明显不行。下面通过flex技术来实现这个功能。思路：以告警数据为例，告警数据通过mq、socket等方式进入系统，在java服务器通过监听来监听消息队列数据，接收后推送到flex各个客户端

前言：
对于数据实时性页面呈现非常强的业务，通过前台定时刷新这样的效率明显不行。下面通过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加入推送服务。
<destination id="serverpushserver">
<properties>
<source>
com.east.flex.serverpush.chart.ServerPushServer
</source>
</properties>
</destination> 1.服务发布 Java代码：ServerPushServer.Java

Java代码?

flex?<wbr>blazeds推送机制

package?com.east.flex.serverpush.chart; ??
??
??
public?class?ServerPushServer?{????
????//ServerPushThread?serverpushthread?=?new??ServerPushThread();???
????MessageBrokFactory?messageBrokFactory; ??
????public?void?controlThread(String?str){ ??
????????if(str.equals("start")){ ??
????????????try{ ??
????????????????messageBrokFactory=new?MessageBrokFactory(); ??
????????????????messageBrokFactory.start(); ??
????????????}catch?(Exception?e){ ??
?????????????????e.printStackTrace(); ??
????????????} ??
???????????????
????????//serverpushthread.start();???
????????}else{ ??
????????????messageBrokFactory.stop(); ??
????????????//serverpushthread.stop();???
????????} ??
????} ??
}??

2.消息处理事务: messageBrokFactory
具有功能：启动mq,并监听mq接收消息,消息推送到flex.三大功能。

flex?<wbr>blazeds推送机制

package?com.east.flex.serverpush.chart; ??
??
import?javax.jms.JMSException; ??
??
import?javax.jms.Message; ??
import?javax.jms.MessageListener; ??
import?javax.jms.TextMessage; ??
import?javax.naming.NamingException; ??
??
import?org.apache.log4j.Logger; ??
??
import?com.zznode.gixpert.provider.activemq.service.impl.ActiveMQAdapter; ??
??
import?flex.messaging.MessageBroker; ??
import?flex.messaging.messages.AsyncMessage; ??
import?flex.messaging.util.UUIDUtils; ??
??
??
??
public?class?MessageBrokFactory?{ ??
??????
????private?MessageBroker?msgBroker; ??
??????
????private?String?clientId?; ??
???????
??????
??????
????private??static?int?port=61616; ??
??????
????private?static?String?hostname?=?"localhost"; ??
??????
????private??static?String?user?=?""; ??
??????
????private??static?String?password=?""; ??
??????
????private?String?queuename="sendToFlex"; ??
??????
????ActiveMQAdapter?amqAdapter; ??
??????
????private?static?boolean??pattern; ??
??????
????private??static?String?durable="是"; ??
??????
????private??static?String?sleepTime="否"; ??
??????
????private??static?String?receiveTimeOut="否"; ??
??????
???????
????private?Logger?logger?=?Logger.getLogger(ActiveMQAdapter.class); ??
??
????public?boolean?sedMessage(String?sub,Object?message)?{ ??
????????try{ ??
????????????AsyncMessage?msg?=?new?AsyncMessage(); ??
????????????msg.setDestination("serverpush"); ??
????????????msg.setClientId(clientId); ??
????????????msg.setMessageId(UUIDUtils.createUUID()); ??
????????????msg.setTimestamp(System.currentTimeMillis()); ??
??????????????
????????????msg.setBody(message); ??
????????????msgBroker.routeMessageToService(msg,?null); ??
????????????return?true; ??
????????}catch(Exception?e){ ??
????????????e.printStackTrace(); ??
????????????return?true; ??
????????} ??
???????????
????} ??
??????
????public?void?start(){ ??
????????try?{ ??
????????????//初始化flex消息组建???
????????????msgBroker?=?MessageBroker.getMessageBroker(null); ??
????????????clientId?=?UUIDUtils.createUUID(); ??
????????????//启动mq消息接收???
????????????amqAdapter?=?new?ActiveMQAdapter(); ??
????????????//版本1.0只需要对hostname??port?起作用?有更高需求再另添加???
????????????amqAdapter.setUrl("failover:(tcp://"+hostname+":"+port+"?wireFormat.maxInactivityDuration=0)"); ??
???????????
????????????amqAdapter.addMessageListener(queuename,?pattern,?new?MyTestMessageListener()); ??
????????}?catch?(JMSException?e)?{ ??
????????????String?msg?=?"Cannot?start?AlarmCollectQueue."; ??
????????????logger.error(msg,?e); ??
????????????System.out.println("set?config?ex"+e); ??
????????????e.printStackTrace(); ??
????????}?catch?(NamingException?e)?{ ??
????????????String?msg?=?"Naming?context?error."; ??
????????????logger.error(msg,?e); ??
????????????System.out.println("set?config?ex"+e); ??
????????????e.printStackTrace(); ??
????????} ??
????} ??
??????
????public?void?stop()?{ ??
????????amqAdapter.close(); ??
????} ??
??????
????class?MyTestMessageListener?implements?MessageListener?{ ??
???????????
????????public?void?onMessage(Message?msg)?{ ??
????????????try{ ??
????????????????//???
????????????????String?message=((TextMessage)msg).getText(); ??
????????????????logger.info("recieve?a?message?from:?"+msg.getJMSDestination()+"??"?+ ??
????????????????????????"message?text:?"+message); ??
????????????????sedMessage("test",message); ??
????????????}catch(Exception?e){ ??
????????????????e.printStackTrace(); ??
????????????????System.out.println("set?config?ex"+e); ??
????????????} ??
????????} ??
????} ??
??
}??

2.ActiveMq接收消息的代码见附件工程。在此不列出。
Flex端：本例以一个textarea文本来显示数据。告警数据具体可以使用datagrid
展现代码getMessageClient.mxml

Flex代码?

flex?<wbr>blazeds推送机制

<DIV?class=quote_title>引用</DIV><DIV?class=quote_div><PRE?class=java?name="code"><PRE?class=java?name="code"><PRE?class=java?name="code"><?xml?version="1.0"?encoding="utf-8"?> ??
<mx:Application?xmlns:mx="http://www.adobe.com/2006/mxml"?layout="absolute"> ??
?????<mx:Script> ??
????????<![CDATA[ ??
????????????import?com.east.flex.entity.Product; ??
????????import?mx.messaging.messages.IMessage; ??
????????import?mx.messaging.events.MessageEvent; ??
????????import?mx.utils.ArrayUtil; ??
????????import?mx.rpc.events.ResultEvent; ??
????????import?mx.collections.ArrayCollection; ??
????????import?mx.controls.Alert; ??
????????import?mx.messaging.ChannelSet; ??
????????import?mx.messaging.Consumer; ??
??
????????[Bindable] ??
????????private?var?p:Product; ??
???????????
????????private?var?array:ArrayUtil; ??
???????????
????????[Bindable] ??
????????private?var?expensesAC:ArrayCollection?=?null; ??
???????????????
????????public?var?consumer2:Consumer?=?null; ??
??????????????
????????public?function?messageHandler(event:MessageEvent):void ??
????????{ ??
????????????textArea.text?=??event.message.body+textArea.text; ??
????????????trace("server?push?data?to?client.............."); ??
????????} ??
???????????
???????????
????????public?function?remoteServerCallback(event:ResultEvent):void ??
????????{ ??
????????????trace("Server?is?start...."); ??
????????????//Alert.show("Server?is?start....."+event.message); ??
????????} ??
???????????
????????public?function?startServer(event:MouseEvent):void ??
????????{ ??
//??????????server.start(); ??
????????????server.controlThread("start"); ??
????????} ??
???????????
????????public?function?stopServer(event:MouseEvent):void ??
????????{ ??
//??????????server.stop(); ??
????????????server.controlThread("stop"); ??
????????} ??
????????]]> ??
????</mx:Script> ??
???????
????<mx:Consumer?id="consumer"?destination="serverpush"?message="messageHandler(event)"/> ??
???????
????<mx:RemoteObject?id="server"?destination="serverpushserver"??result="remoteServerCallback(event)"/> ??
???????
???? ??
????<mx:SolidColor?id="sc1"?color="blue"?alpha=".3"/> ??
????<mx:SolidColor?id="sc2"?color="red"?alpha=".3"/> ??
????<mx:SolidColor?id="sc3"?color="green"?alpha=".3"/> ??
??
???? ??
????<mx:Stroke?id?=?"s1"?color="blue"?weight="2"/> ??
????<mx:Stroke?id?=?"s2"?color="red"?weight="2"/> ??
????<mx:Stroke?id?=?"s3"?color="green"?weight="2"/> ??
??
????<mx:Panel?title="LineChart?and?AreaChart?Controls?Example"????
????????height="100%"?width="100%"?layout="absolute"> ??
??
????????????<mx:TextArea?id="textArea"?width="1017"?height="565"> ??
???????????????????
????????????</mx:TextArea> ??
??
???????????
???????????
????????<mx:Button?id="autoOk"?label="AutoRefresh"?x="1038"?y="87"?width="107"?click="consumer.subscribe()"?enabled="{!consumer.subscribed}"/> ??
????????<mx:Button?id="cancel"?x="1038"?y="135"?width="107"?label="Cancel?Auto?"??click="consumer.unsubscribe()"?enabled="{consumer.subscribed}"/> ??
????????<mx:Button?id="cancel0"?x="1038"?y="187"?width="107"?label="Start?Server"??click="startServer(event)"/> ??
????????<mx:Button?id="cancel1"?x="1038"?y="237"?width="107"?label="Stop?Server"??click="stopServer(event)"/> ??
????</mx:Panel> ??
</mx:Application></PRE></PRE></PRE></DIV>??

引用

Java代码?

flex?<wbr>blazeds推送机制

<PRE?class=java?name="code"><PRE?class=java?name="code"><?xml?version="1.0"?encoding="utf-8"?> ??
<mx:Application?xmlns:mx="http://www.adobe.com/2006/mxml"?layout="absolute"> ??
?????<mx:Script> ??
????????<![CDATA[ ??
????????????import?com.east.flex.entity.Product; ??
????????import?mx.messaging.messages.IMessage; ??
????????import?mx.messaging.events.MessageEvent; ??
????????import?mx.utils.ArrayUtil; ??
????????import?mx.rpc.events.ResultEvent; ??
????????import?mx.collections.ArrayCollection; ??
????????import?mx.controls.Alert; ??
????????import?mx.messaging.ChannelSet; ??
????????import?mx.messaging.Consumer; ??
??
????????[Bindable] ??
????????private?var?p:Product; ??
???????????
????????private?var?array:ArrayUtil; ??
???????????
????????[Bindable] ??
????????private?var?expensesAC:ArrayCollection?=?null; ??
???????????????
????????public?var?consumer2:Consumer?=?null; ??
??????????????
????????public?function?messageHandler(event:MessageEvent):void??
????????{ ??
????????????textArea.text?=??event.message.body+textArea.text; ??
????????????trace("server?push?data?to?client.............."); ??
????????} ??
???????????
???????????
????????public?function?remoteServerCallback(event:ResultEvent):void??
????????{ ??
????????????trace("Server?is?start...."); ??
????????????//Alert.show("Server?is?start....."+event.message);???
????????} ??
???????????
????????public?function?startServer(event:MouseEvent):void??
????????{ ??
//??????????server.start();???
????????????server.controlThread("start"); ??
????????} ??
???????????
????????public?function?stopServer(event:MouseEvent):void??
????????{ ??
//??????????server.stop();???
????????????server.controlThread("stop"); ??
????????} ??
????????]]> ??
????</mx:Script> ??
???????
????<mx:Consumer?id="consumer"?destination="serverpush"?message="messageHandler(event)"/> ??
???????
????<mx:RemoteObject?id="server"?destination="serverpushserver"??result="remoteServerCallback(event)"/> ??
???????
???? ??
????<mx:SolidColor?id="sc1"?color="blue"?alpha=".3"/> ??
????<mx:SolidColor?id="sc2"?color="red"?alpha=".3"/> ??
????<mx:SolidColor?id="sc3"?color="green"?alpha=".3"/> ??
??
???? ??
????<mx:Stroke?id?=?"s1"?color="blue"?weight="2"/> ??
????<mx:Stroke?id?=?"s2"?color="red"?weight="2"/> ??
????<mx:Stroke?id?=?"s3"?color="green"?weight="2"/> ??
??
????<mx:Panel?title="LineChart?and?AreaChart?Controls?Example"????
????????height="100%"?width="100%"?layout="absolute"> ??
??
????????????<mx:TextArea?id="textArea"?width="1017"?height="565"> ??
???????????????????
????????????</mx:TextArea> ??
??
???????????
???????????
????????<mx:Button?id="autoOk"?label="AutoRefresh"?x="1038"?y="87"?width="107"?click="consumer.subscribe()"?enabled="{!consumer.subscribed}"/> ??
????????<mx:Button?id="cancel"?x="1038"?y="135"?width="107"?label="Cancel?Auto?"??click="consumer.unsubscribe()"?enabled="{consumer.subscribed}"/> ??
????????<mx:Button?id="cancel0"?x="1038"?y="187"?width="107"?label="Start?Server"??click="startServer(event)"/> ??
????????<mx:Button?id="cancel1"?x="1038"?y="237"?width="107"?label="Stop?Server"??click="stopServer(event)"/> ??
????</mx:Panel> ??
</mx:Application></PRE></PRE>??

Java代码?

flex?<wbr>blazeds推送机制

<PRE?class=java?name="code"><?xml?version="1.0"?encoding="utf-8"?> ??
<mx:Application?xmlns:mx="http://www.adobe.com/2006/mxml"?layout="absolute"> ??
?????<mx:Script> ??
????????<![CDATA[ ??
????????????import?com.east.flex.entity.Product; ??
????????import?mx.messaging.messages.IMessage; ??
????????import?mx.messaging.events.MessageEvent; ??
????????import?mx.utils.ArrayUtil; ??
????????import?mx.rpc.events.ResultEvent; ??
????????import?mx.collections.ArrayCollection; ??
????????import?mx.controls.Alert; ??
????????import?mx.messaging.ChannelSet; ??
????????import?mx.messaging.Consumer; ??
??
????????[Bindable] ??
????????private?var?p:Product; ??
???????????
????????private?var?array:ArrayUtil; ??
???????????
????????[Bindable] ??
????????private?var?expensesAC:ArrayCollection?=?null; ??
???????????????
????????public?var?consumer2:Consumer?=?null; ??
??????????????
????????public?function?messageHandler(event:MessageEvent):void??
????????{ ??
????????????textArea.text?=??event.message.body+textArea.text; ??
????????????trace("server?push?data?to?client.............."); ??
????????} ??
???????????
???????????
????????public?function?remoteServerCallback(event:ResultEvent):void??
????????{ ??
????????????trace("Server?is?start...."); ??
????????????//Alert.show("Server?is?start....."+event.message);???
????????} ??
???????????
????????public?function?startServer(event:MouseEvent):void??
????????{ ??
//??????????server.start();???
????????????server.controlThread("start"); ??
????????} ??
???????????
????????public?function?stopServer(event:MouseEvent):void??
????????{ ??
//??????????server.stop();???
????????????server.controlThread("stop"); ??
????????} ??
????????]]> ??
????</mx:Script> ??
???????
????<mx:Consumer?id="consumer"?destination="serverpush"?message="messageHandler(event)"/> ??
???????
????<mx:RemoteObject?id="server"?destination="serverpushserver"??result="remoteServerCallback(event)"/> ??
???????
???? ??
????<mx:SolidColor?id="sc1"?color="blue"?alpha=".3"/> ??
????<mx:SolidColor?id="sc2"?color="red"?alpha=".3"/> ??
????<mx:SolidColor?id="sc3"?color="green"?alpha=".3"/> ??
??
???? ??
????<mx:Stroke?id?=?"s1"?color="blue"?weight="2"/> ??
????<mx:Stroke?id?=?"s2"?color="red"?weight="2"/> ??
????<mx:Stroke?id?=?"s3"?color="green"?weight="2"/> ??
??
????<mx:Panel?title="LineChart?and?AreaChart?Controls?Example"????
????????height="100%"?width="100%"?layout="absolute"> ??
??
????????????<mx:TextArea?id="textArea"?width="1017"?height="565"> ??
???????????????????
????????????</mx:TextArea> ??
??
???????????
???????????
????????<mx:Button?id="autoOk"?label="AutoRefresh"?x="1038"?y="87"?width="107"?click="consumer.subscribe()"?enabled="{!consumer.subscribed}"/> ??
????????<mx:Button?id="cancel"?x="1038"?y="135"?width="107"?label="Cancel?Auto?"??click="consumer.unsubscribe()"?enabled="{consumer.subscribed}"/> ??
????????<mx:Button?id="cancel0"?x="1038"?y="187"?width="107"?label="Start?Server"??click="startServer(event)"/> ??
????????<mx:Button?id="cancel1"?x="1038"?y="237"?width="107"?label="Stop?Server"??click="stopServer(event)"/> ??
????</mx:Panel> ??
</mx:Application></PRE>??

Java代码?

flex?<wbr>blazeds推送机制

<?xml?version="1.0"?encoding="utf-8"?> ??
<mx:Application?xmlns:mx="http://www.adobe.com/2006/mxml"?layout="absolute"> ??
?????<mx:Script> ??
????????<![CDATA[ ??
????????????import?com.east.flex.entity.Product; ??
????????import?mx.messaging.messages.IMessage; ??
????????import?mx.messaging.events.MessageEvent; ??
????????import?mx.utils.ArrayUtil; ??
????????import?mx.rpc.events.ResultEvent; ??
????????import?mx.collections.ArrayCollection; ??
????????import?mx.controls.Alert; ??
????????import?mx.messaging.ChannelSet; ??
????????import?mx.messaging.Consumer; ??
??
????????[Bindable] ??
????????private?var?p:Product; ??
???????????
????????private?var?array:ArrayUtil; ??
???????????
????????[Bindable] ??
????????private?var?expensesAC:ArrayCollection?=?null; ??
???????????????
????????public?var?consumer2:Consumer?=?null; ??
??????????????
????????public?function?messageHandler(event:MessageEvent):void??
????????{ ??
????????????textArea.text?=??event.message.body+textArea.text; ??
????????????trace("server?push?data?to?client.............."); ??
????????} ??
???????????
???????????
????????public?function?remoteServerCallback(event:ResultEvent):void??
????????{ ??
????????????trace("Server?is?start...."); ??
????????????//Alert.show("Server?is?start....."+event.message);???
????????} ??
???????????
????????public?function?startServer(event:MouseEvent):void??
????????{ ??
//??????????server.start();???
????????????server.controlThread("start"); ??
????????} ??
???????????
????????public?function?stopServer(event:MouseEvent):void??
????????{ ??
//??????????server.stop();???
????????????server.controlThread("stop"); ??
????????} ??
????????]]> ??
????</mx:Script> ??
???????
????<mx:Consumer?id="consumer"?destination="serverpush"?message="messageHandler(event)"/> ??
???????
????<mx:RemoteObject?id="server"?destination="serverpushserver"??result="remoteServerCallback(event)"/> ??
???????
???? ??
????<mx:SolidColor?id="sc1"?color="blue"?alpha=".3"/> ??
????<mx:SolidColor?id="sc2"?color="red"?alpha=".3"/> ??
????<mx:SolidColor?id="sc3"?color="green"?alpha=".3"/> ??
??
???? ??
????<mx:Stroke?id?=?"s1"?color="blue"?weight="2"/> ??
????<mx:Stroke?id?=?"s2"?color="red"?weight="2"/> ??
????<mx:Stroke?id?=?"s3"?color="green"?weight="2"/> ??
??
????<mx:Panel?title="LineChart?and?AreaChart?Controls?Example"????
????????height="100%"?width="100%"?layout="absolute"> ??
??
????????????<mx:TextArea?id="textArea"?width="1017"?height="565"> ??
???????????????????
????????????</mx:TextArea> ??
??
???????????
???????????
????????<mx:Button?id="autoOk"?label="AutoRefresh"?x="1038"?y="87"?width="107"?click="consumer.subscribe()"?enabled="{!consumer.subscribed}"/> ??
????????<mx:Button?id="cancel"?x="1038"?y="135"?width="107"?label="Cancel?Auto?"??click="consumer.unsubscribe()"?enabled="{consumer.subscribed}"/> ??
????????<mx:Button?id="cancel0"?x="1038"?y="187"?width="107"?label="Start?Server"??click="startServer(event)"/> ??
????????<mx:Button?id="cancel1"?x="1038"?y="237"?width="107"?label="Stop?Server"??click="stopServer(event)"/> ??
????</mx:Panel> ??
</mx:Application>??

使用操作：
在第一次打开的页面点击startserver.这个共是启动监听，注意只需要在第一次打开点击了就行，其他打开的页面不需要。再次点击
然后点击autorefesh，用来启动自动显示数据，注意autorefesh这不是自动刷新请求数据。是推动上来的数据。
其他打开的页面都只要点击autorefesh即可。
启动一个acitmvemq在提供的页面发送数据，可以一次发送几千条数据。同事打开几个页面来测试下下过。
[color=red]优化：[/color]
项目最近开始运行发现客户端web非常耗费cpu,非常头疼，当然这与我的数据量肯定是有关系的，我后台推送到前台的数据大约每秒100条。消息量少时streaming没有问题。
后来经过了多种情况的测试，最后发现把streaming设置为polling方式轮询推送占用的cpu资源一下子降了下来。后来吧轮询时间这只为0.3s效果比streaming还好。streaming方式是，消息上来的太快，表格显示时非常闪眼睛，用户没法操作，相反设为polling后这种现象降低了。
总结steaming耗费cpu大的原因是，flex执行死循环时比java执行死循环耗资源打多了。也就是对于一直占用cpu不释放。steaming方式消息量大了也跟执行死循环一样，一直在执行不释放cpu.
下面是改造后的配置
Java代码?

flex?<wbr>blazeds推送机制

<channel-definition?id="my-polling-amf"?class="mx.messaging.channels.AMFChannel"> ??
???????????<endpoint?url="http://{server.name}:{server.port}/{context.root}/messagebroker/amfpolling"?class="flex.messaging.endpoints.AMFEndpoint"/> ??
???????????<properties> ??
???????????????<polling-enabled>true</polling-enabled> ??
???????????????<polling-interval-seconds>0.2</polling-interval-seconds> ??
?????????????????
???????????</properties> ??
???????</channel-definition>??

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!