Flex与Java的简易消息推送机制举例-

Flex与Java的简易消息推送机制举例

使用adobe提供的java flex通信包blazeDS中提供的推送功能实现Java到Flex的数据推送过程

1. 从adobe官网下载blazeds.war,这是一个压缩包,里面的包含META-INF和WEB-INF两个文件夹,目录结构如下:其中含META-INF仅包含MANIFEST.MF文件, WEB-INF中包含classes、flex、lib及src四个文件夹，其中flex文件夹中包含四个配置文件，lib中包含java和flex通讯所需要的jar包。

Flex文件夹中包含：

Lib文件夹包含：

2. 创建Java后台的推送服务项目

新建一个DynamicWeb Project,如:JavaPushServer

点击Next

点击Next

点击Finish,这时候创建了一个空的动态web项目,使用Tomcat6运行.这时候Eclipse中的见到的项目信息如下:Java Resources中的src为java源代码,Libraries中为java运行所依赖的jar包,如果已经配置好了Tomcat运行环境,则Apache Tomcat v6.0…目录下自动加载了Tomcat的jar包,和JRE System Library目录下的jar包

因为通讯需要blazeds,所以,这时候,可以将blazeds.war解压出来的两个文件夹替换掉刚才创建的web工程中WebContent文件目录下的内容.替换之后可以看到WebContent文件夹下内容变化情况:

以及Java Sources中的Libraries中Web App Libraries中增加了很多jar包,这些就是blazeds.war中所包含的jar包

引入blazeds.war之后,开始写Java代码:

创建名为Tick的java实体类:

package com.eunut.vo;

//实体类

public class Tick {

privateString seqno;

publicString getSeqno(){

returnseqno;

}

//字符类型,为测试用仅仅加了一个属性,可以为多个

publicvoid setSeqno(String seqno)

{

this.seqno=seqno;

}

}

然后再创建一个名为TickCacheServlet的Servlet

举例代码如下:

import java.io.IOException;

importjavax.servlet.ServletException;

importjavax.servlet.http.HttpServlet;

importjavax.servlet.http.HttpServletRequest;

importjavax.servlet.http.HttpServletResponse;

import com.eunut.vo.Tick;

importflex.messaging.MessageBroker;

importflex.messaging.messages.AsyncMessage;

importflex.messaging.util.UUIDUtils;

/**

* Servlet implementation classTickCacheServlet

*/

public class TickCacheServletextends HttpServlet {

privatestatic final long serialVersionUID = 1L;

//线程

privatestatic FeedThread thread;

/**

* @see HttpServlet#HttpServlet()

*/

public TickCacheServlet() {

super();

// TODO Auto-generated constructor stub

}

/**

* @see HttpServlet#doGet(HttpServletRequestrequest, HttpServletResponse response)

*/

protectedvoid doGet(HttpServletRequest request, HttpServletResponse response) throwsServletException, IOException {

//TODO Auto-generated method stub

Stringcmd=request.getParameter("cmd");

if(cmd.equals("start"))

{

start();

}

if(cmd.equals("stop"))

{

stop();

}

}

/**

* @see HttpServlet#doPost(HttpServletRequestrequest, HttpServletResponse response)

*/

protectedvoid doPost(HttpServletRequest request, HttpServletResponse response) throwsServletException, IOException {

//TODO Auto-generated method stub

super.doPost(request,response);

}

//初始化

publicvoid init() throws ServletException{

super.init();

}

publicvoid destroy(){

super.destroy();

}

//启动监听

publicvoid start(){

if(thread==null)

{

thread=newFeedThread();

thread.start();

}

System.out.println("start!!!");

}

//停止监听

publicvoid stop(){

thread.running=false;

thread=null;

}

//循环发送消息进程

publicstatic class FeedThread extends Thread

{

publicboolean running=true;

publicvoid run(){

//总是无法获取msgBroker,web.xml需要有MessageBrokerServlet的配置信息

MessageBrokermsgBroker=MessageBroker.getMessageBroker(null);

StringclientID=UUIDUtils.createUUID();

inti=0;

while(running){

Ticktick=new Tick();

tick.setSeqno(String.valueOf(i));

System.out.println(i);

AsyncMessagemsg=new AsyncMessage();

msg.setDestination("tick-data-feed");

msg.setHeader(AsyncMessage.SUBTOPIC_HEADER_NAME,"tick");

msg.setClientId(clientID);

msg.setMessageId(UUIDUtils.createUUID());

msg.setTimestamp(System.currentTimeMillis());

msg.setBody(tick);

msgBroker.routeMessageToService(msg,null);

i++;

try

{

Thread.sleep(2000);

}

catch(InterruptedExceptione)

{

}

}

}

}

}

3. 修改Java后台的推送服务项目的配置文件:

WebContent\WEB-INF\messaging-config.xml

<?xml version="1.0"encoding="UTF-8"?>

<serviceid="message-service"

class="flex.messaging.services.MessageService">

<adapters>

<adapter-definitionid="actionscript"class="flex.messaging.services.messaging.adapters.ActionScriptAdapter"default="true" />

<!-- <adapter-definitionid="jms"class="flex.messaging.services.messaging.adapters.JMSAdapter"/>-->

</adapters>

<default-channels>

<channelref="my-polling-amf"/>

</default-channels>

<!-- 增加配置信息 -->

<destination id="tick-data-feed">

<properties>

<server>

<allow-subtopics>true</allow-subtopics>

<subtopic-separator>.</subtopic-separator>

</server>

</properties>

<channels>

<channel ref="my-polling-amf" />

<channel ref="my-streaming-amf" />

</channels>

</destination>

</service>

以及services-config.xml中补充:

<!-- 补充my-streaming-amf通道配置信息 -->

<channel-definitionid="my-streaming-amf"class="mx.messaging.channels.StreamingAMFChannel">

<endpointurl="http://{server.name}:{server.port}/{context.root}/messagebroker/streamingamf"class="flex.messaging.endpoints.StreamingAMFEndpoint"/>

<properties>

<idle-timeout-minutes>0</idle-timeout-minutes>

<max-streaming-clients>10</max-streaming-clients>

<server-to-client-heartbeat-millis>5000</server-to-client-heartbeat-millis>

<user-agent-settings>

<user-agent match-on="MSIE"kickstart-bytes="2048"max-streaming-connections-per-session="1"/>

<user-agentmatch-on="Firefox" kickstart-bytes="2048"max-streaming-connections-per-session="1"/>

</user-agent-settings>

</properties>

</channel-definition>

同时,还要确保web.xml中包含下面的内容:

<!-- MessageBroker Servlet-必须的,否则获取不到MessageBrokerServlet,无法发送消息 -->

<servlet>

<servlet-name>MessageBrokerServlet</servlet-name>

<display-name>MessageBrokerServlet</display-name>

<servlet-class>flex.messaging.MessageBrokerServlet</servlet-class>

<init-param>

<param-name>services.configuration.file</param-name>

<param-value>/WEB-INF/flex/services-config.xml</param-value>

</init-param>

<load-on-startup>1</load-on-startup>

</servlet>

<servlet-mapping>

<servlet-name>MessageBrokerServlet</servlet-name>

<url-pattern>/messagebroker/*</url-pattern>

</servlet-mapping>

<!-- 下面的是自动生成的 -->

<servlet>

<javaee:description></javaee:description>

<javaee:display-name>TickCacheServlet</javaee:display-name>

<servlet-name>TickCacheServlet</servlet-name>

<servlet-class>TickCacheServlet</servlet-class>

</servlet>

<servlet-mapping>

<servlet-name>TickCacheServlet</servlet-name>

<url-pattern>/TickCacheServlet</url-pattern>

</servlet-mapping>

4. 创建Flex客户端应用程序FlexPushClient.mxml:

<?xml version="1.0"encoding="utf-8"?>

<s:Applicationxmlns:fx="http://ns.adobe.com/mxml/2009"

xmlns:s="library://ns.adobe.com/flex/spark"

xmlns:mx="library://ns.adobe.com/flex/mx"minWidth="955" minHeight="600">

<fx:Declarations>

<!--将非可视元素（例如服务、值对象）放在此处 -->

</fx:Declarations>

<fx:Script>

<![CDATA[

importcom.eunut.Tick;

importmx.controls.Alert;

importmx.messaging.ChannelSet;

importmx.messaging.Consumer;

importmx.messaging.channels.StreamingAMFChannel;

importmx.messaging.events.ChannelEvent;

importmx.messaging.events.MessageEvent;

[Bindable]

publicvar tick:Tick;

publicfunction submsg():void

{

varconsumer:Consumer=new Consumer();

consumer.destination="tick-data-feed";

consumer.subtopic="tick";

consumer.addEventListener(MessageEvent.MESSAGE,messageHandler);

varurl:String ="http://localhost:8080/JavaPushServer/";

varmyStreamingAMF:StreamingAMFChannel = newStreamingAMFChannel(url+"my-streaming-amf",url+"messagebroker/streamingamf");

varchannelSet:ChannelSet = new ChannelSet();

channelSet.addChannel(myStreamingAMF);

consumer.channelSet= channelSet;

// consumer.channelSet = newChannelSet(["my-streaming-amf"]);

consumer.subscribe(); //开始接收

Alert.show("消费者初始化完毕！");

}

privatefunction messageHandler(event:MessageEvent):void

{

vartick:Tick = event.message.body as Tick;

txtTick.text= tick.seqno;

}

]]>

</fx:Script>

<mx:Panelx="24" y="35" width="362" height="302"layout="absolute" title="Watch Tick">

<mx:Labelx="72" y="43" text="Label"id="txtTick"/>

<mx:Buttonx="132" y="41" label="Button"click="submsg()"/>

</mx:Panel>

</s:Application>

还要创建Tick的as类:

package com.eunut

{

importmx.rpc.remoting.RemoteObject;

[RemoteClass(alias="com.eunut.vo.Tick")] //用于和java后台类转换alias为java类的命名空间,否则不能转换

[Bindable]

publicclass Tick

{

publicfunction Tick()

{

}

publicvar seqno:String;

}

}

配置此flex的应用程序的服务器为刚才创建好的JavaPushServer

好了,这样就可以了哦,现在就是如果看到推送的效果了,启动方式如下:

首先运行发布到Tomcat上的JavaPushServer,然后,打开客户端网址: http://localhost:8080/JavaPushServer/FlexPushClient.html
这时候显示的界面如下:

此时,点击Button按钮,启动监听程序.

然后,需要启动服务器端的监听:

http://localhost:8080/JavaPushServer/TickCacheServlet?cmd=start

如果一切顺利,则可以看到Label显示的数字从0开始逐个增加.

通过调用:

http://localhost:8080/JavaPushServer/TickCacheServlet?cmd=stop

就关闭服务器端的监听程序

注意:途中可能出现这样那样的问题,绝大部分都是因为配置文件不正确引起的,所以,遇到问题时,一定要仔细查看配置文件中的内容是否正确.