Flex采用blazeds实现服务器向（特定标识的）客户端推数据(基于consumer模式)

前言：

有很多类似股票、外汇、期货等实时行情这种应用，客户端需要显示行情牌价等信息。

目前的作法是：客户端定时向服务器请求，无论数据是否有更新，都把数据发到客户端。

我们这里讲的一种技术不同以上这个做法，我们是采用服务器向客户端推的这种方式，该方式的好处不言自明。blazeds中有一个名为：StreamingAMFChannel　的通道，我们就是采用它来实现向客户端推这个功能。

Tick.java :

package test;

import java.math.BigDecimal;

import java.util.Date;

public class Tick {

private BigDecimal askPrice;

private BigDecimal bidPrice;

private BigDecimal midPrice;

private Date tickTime;

private String seqno;

public String getSeqno() {

return seqno;

}

public void setSeqno(String seqno) {

this.seqno = seqno;

}

public BigDecimal getAskPrice() {

return askPrice;

}

public void setAskPrice(BigDecimal askPrice) {

this.askPrice = askPrice;

}

public BigDecimal getBidPrice() {

return bidPrice;

}

public void setBidPrice(BigDecimal bidPrice) {

this.bidPrice = bidPrice;

}

public BigDecimal getMidPrice() {

return midPrice;

}

public void setMidPrice(BigDecimal midPrice) {

this.midPrice = midPrice;

}

public Date getTickTime() {

return tickTime;

}

public void setTickTime(Date tickTime) {

this.tickTime = tickTime;

}

}

FeedThread .java :

public static class FeedThread extends Thread {

private String uid;

public void setUid(String uid) {

this.uid = uid;

}

public FeedThread(String uid){

this.uid = uid;

}

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.setAskPrice(new BigDecimal("100"));

tick.setBidPrice(new BigDecimal("100"));

tick.setMidPrice(new BigDecimal("100"));

tick.setTickTime(new Date());

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

System.out.println(i+"--"+uid);

AsyncMessage msg = new AsyncMessage();

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

msg.setHeader("DSSubtopic", "tick"+uid);

msg.setClientId(clientID);

msg.setMessageId(UUIDUtils.createUUID());

msg.setTimestamp(System.currentTimeMillis());

msg.setBody(tick);

msgBroker.routeMessageToService(msg, null);

i++;

try {

Thread.sleep(20);

} catch (InterruptedException e) {

}

}

}

}

UserService.java :

package sws.service;

import module.User;

public interface UserService {

public void Test(String uid);

public void stop(String uid);

}

UserServiceImpl.java :

package sws.service;

import java.util.HashMap;

import java.util.Map;

import module.User;

import org.springframework.beans.factory.annotation.Autowired;

import sws.dao.UserDao;

import test.TickServlet.FeedThread;

public class UserServiceImpl implements UserService {

private static Map<String,Thread> map = new HashMap<String,Thread>();

public void Test(String uid) {

Boolean b = map.containsKey(uid);

if(!b){

FeedThread thread = new FeedThread(uid);

thread.start();

map.put(uid, thread);

}

}

public void stop(String uid) {

FeedThread thread = (FeedThread)map.get(uid);

thread.running = false;

}

}

下一步加入flex配置文件。

flex的配置文件默认有四个，文件目录在WebContent\WEB-INF\flex目录下。

messaging-config.xml

proxy-config.xml

remoting-config.xml

services-config.xml

其实，这中间用的是一个，就是services-config.xml，只是在services-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>

Tick.as文件内容如下：

//Tick.as

package

{

[RemoteClass(alias="test.Tick")]

[Bindable]

public class Tick

{

public var askPrice:Number;

public var bidPrice:Number;

public var midPrice:Number;

public var tickTime:Date;;

public var seqno:String;

}

}

再在main.mxml文件中，加入如下代码：

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

<mx:Application xmlns:mx="http://www.adobe.com/2006/mxml" layout="absolute" viewSourceURL="srcview/index.html" height="378" width="426">

<mx:Script>

<![CDATA[

import mx.controls.Alert;

import mx.rpc.events.ResultEvent;

import mx.rpc.events.FaultEvent;

import mx.messaging.Consumer;

import mx.messaging.Channel;

import mx.messaging.ChannelSet;

import mx.messaging.events.MessageEvent;

[Bindable]

public var tick:Tick;

public function postUrl():void {

//

// httpservice.url = "http://localhost:8080/test/servlet/TickServlet?cmd=start&uid="+telId.text;

// httpservice.send();

// httpservice.addEventListener(ResultEvent.RESULT,resultHandler);

// httpservice.addEventListener(FaultEvent.FAULT,faultHandler);

user.Test(telId.text);

}

public function resultHandler(event:ResultEvent):void{

submsg();

}

public function faultHandler(event:FaultEvent):void{

trace(event.fault);

}

public function submsg():void

{

var consumer:Consumer = new Consumer();

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

consumer.subtopic = "tick"+telId.text;

consumer.channelSet = new ChannelSet(["my-streaming-amf"]);

consumer.addEventListener(MessageEvent.MESSAGE, messageHandler);

consumer.subscribe();

}

private function messageHandler(event:MessageEvent):void

{

tick = event.message.body as Tick;

txtTick.text = tick.seqno;

}

private function stop():void {

user.stop(telId.text);

}

]]>

</mx:Script>

//RemoteObject 的方式远程调用java类 userService在rremoting-config.xml中配置

<mx:RemoteObject id="user" destination="userService" source="userService" result="resultHandler(event)"/>

<mx:HTTPService id="httpservice" />

<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="41" label="start" click="postUrl(); "/>

<mx:TextInput id="telId" x="37" y="83"/>//标识用户

<mx:Button label="stop" click="stop();" x="216" y="41"/>

</mx:Panel>

</mx:Application>