flex+blazeds+java后台消息推送(简单示例)

首先，新建一个web项目，添加开发flex需要的jar包，和blazeds相关文件，然后把web项目转成flex项目(不明白怎么弄没事，过几天会写一篇这样的文章)，然后修改一下services-config.xml和messaging-config.xml文件，添加如下代码

services-config.xml

<channel-definition
id="my-streaming-amf"
class="mx.messaging.channels.StreamingAMFChannel">
<endpoint
url="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-agent match-on="Firefox"
kickstart-bytes="2048"
max-streaming-connections-per-session="1"
/>
</user-agent-settings>
</properties>
</channel-definition>

messaging-config.xml

<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>

OK，完了之后，我们新建一个servlet作为服务端

package com.serverpush;

import java.io.IOException;
import
javax.servlet.ServletException;
import
javax.servlet.http.HttpServlet;
import
javax.servlet.http.HttpServletRequest;
import
javax.servlet.http.HttpServletResponse;

import
com.model.Tick;

import flex.messaging.MessageBroker;
import
flex.messaging.messages.AsyncMessage;
import
flex.messaging.util.UUIDUtils;

/**
* Servlet implementation class
TickCacheServlet
*/
public class TickCacheServlet extends HttpServlet
{
private static final long serialVersionUID = 1L;
private static
FeedThread thread;

/**
* @see
HttpServlet#HttpServlet()
*/
public TickCacheServlet() {
super();
//
TODO Auto-generated constructor stub
}

/**
* @see
HttpServlet#doGet(HttpServletRequest request, HttpServletResponse
*
response)
*/
protected void doGet(HttpServletRequest
request,
HttpServletResponse response) throws ServletException, IOException
{
//接收前台参数
String cmd = request.getParameter("cmd");
if
(cmd.equals("start")) {
start();
}
if (cmd.equals("stop"))
{
stop();
}
}

public void start() {
if (thread == null)
{
thread = new
FeedThread();
thread.start();
}
System.out.println("start!!");
}

public
void stop() {
thread.running = false;
thread = null;
}

/**
*
@see HttpServlet#doPost(HttpServletRequest request, HttpServletResponse
*
response)
*/
protected void doPost(HttpServletRequest
request,
HttpServletResponse response) throws ServletException, IOException
{
doGet(request, response);
}
//线程类，每隔多长时间发送一次
public static class
FeedThread extends Thread {
public boolean running = true;
public void
run() {
MessageBroker msgBroker =
MessageBroker.getMessageBroker(null);
String clientID =
UUIDUtils.createUUID();
int i = 0;
while (running) {
Tick tick = new
Tick();
tick.setMessageCN("今天下午6点更新程序需重启服务器");
tick.setSeqNo(String.valueOf(i));
System.out.println(i);

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(tick);
msgBroker.routeMessageToService(msg,
null);
i++;
try {
Thread.sleep(2000);
} catch (InterruptedException
e) {
}
}
}
}
}

在新建一个model用来保存消息

package com.model;

public class Tick {
private String
seqNo;
private String messageCN;

public String getMessageCN()
{
return messageCN;
}

public void setMessageCN(String messageCN)
{
this.messageCN = messageCN;
}

public String getSeqNo()
{
return seqNo;
}

public void setSeqNo(String seqNo)
{
this.seqNo = seqNo;
}
}

后台就搞定了。下面看前台

新建一个mxml文件和一个VO文件

mxml:

<?xml version="1.0"
encoding="utf-8"?>
<s:Application
xmlns: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">
<s:layout>
<s:BasicLayout/>
</s:layout>

<fx:Script>
<![CDATA[
import
mx.controls.Alert;
import mx.messaging.ChannelSet;
import
mx.messaging.Consumer;
import
mx.messaging.events.MessageEvent;

protected function
submsg():void
{
Alert.show("click start");
var consumer:Consumer = new
Consumer();
consumer.destination = "tick-data-feed";
consumer.subtopic =
"tick";
consumer.channelSet = new
ChannelSet(["my-streaming-amf"]);
//添加message的监听，当后台有消息发送时，调用messageHandler

consumer.addEventListener(MessageEvent.MESSAGE, messageHandler);

consumer.subscribe();
}
private function
messageHandler(event:MessageEvent):void
{
var tick:TickVO =
event.message.body as TickVO;
txtTick.text = tick.messageCN+tick.seqNo;

}

]]>
</fx:Script>

<fx:Declarations>
<!--
将非可视元素（例如服务、值对象）放在此处 -->
</fx:Declarations>
<mx:Panel
x="32"
y="43"
width="362"
height="302"
layout="absolute"
title="Watch
Tick">
<mx:Label
x="72"
y="43"
text="Label"
id="txtTick"/>
<mx:Button
x="132"
y="71"
label="Button"
click="submsg()"/>
</mx:Panel>
</s:Application>

VO，就是对应后台的model

package
{
//远程到后台的Tick，也就是说他们俩现在是对应的关系了，你在后台给model赋值了，在前台flex中就可以通过VO
//取得model中的值
[RemoteClass(alias="com.model.Tick")]

[Bindable]
public class TickVO
{
private var
_seqNo:String;
private var _messageCN:String;

public function
TickVO()
{

}
public function get seqNo():String
{
return
_seqNo;
}

public function set seqNo(value:String):void
{
_seqNo
= value;
}

public function get messageCN():String
{
return
_messageCN;
}

public function set
messageCN(value:String):void
{
_messageCN =
value;
}

}
}

ok，下面开始测试：运行mxml，界面如下



点击button，启动监听。

然后在浏览器中输入：http://localhost:8080/serverPush/TickCacheServlet?cmd=start，表示现在我要发消息了，当你运行了这个之后看看你刚才的那个mxml是不是有变化啦