FLEX blazeds推送技术至Flex（2） flex反向推送

参考文章

http://hi.baidu.com/wosinmaiken/blog/item/71f700fdd14aa64fd6887de9.html

这个例子非常经典

在 BlazeDS 3.4 和 BlazeDS 4.0 都可以跑起来

下载链接

http://download.csdn.net/source/2623325

观察者模式的一种实现

订阅者 发布者 模式

客户端是订阅者，服务端是发布者。发布者会向所有订阅者发布相同的信息。

如何区分不同的订阅者。以便实现个性化订阅。

http://topic.csdn.net/u/20100521/17/29d05d17-e3b9-415b-89c1-692b1083072f.html

一种错误的认识

java端：

AsyncMessage msg = new AsyncMessage();

msg.setClientId(entityName);

flex端：

var consumer:Consumer = new Consumer();

consumer.subtopic = entityName;

根据entityname来区分

entityName 可以是 userId

上面这种方法不对

如果某一订阅方只想收到特定的消息，可以在订阅时使用selecor属性来指定过滤条件

经典文章 写的非常经典

Flex + BlazeDS 学习笔记 （一） --- BlazeDS的功能原理及配置实例

http://blog.csdn.net/mebusw/archive/2010/07/17/5742172.aspx

仔细研读 BLazeDS 官方示例 

BlazeDS Test Drive里面示例了主要的功能：

HTTPService

Web Services

Remote-Object

Messaging

消息本身包含消息头和消息体，消息头的格式是固定的，消息体的内容可以随意写。接触过状态机、分布式嵌入式系统编程的会很容易理解。

消息服务允许Flex客户端可以发布(Publish)和订阅(Subscribe)消息，相应的，客户端的被称为生产者(Producer)和消费者(Consumer)，其实也就是发送和接受消息了。

当一个消息被发到已订阅的目的地(destination)时，就会触发一个客户端的message事件。发布的任何消息会发给所有订阅的人，即使是生产者自己。如果某一订阅方只想收到特定的消息，可以在订阅时使用selecor属性来指定过滤条件，只有符合条件的消息才会被服务器转发。如果想要服务器产生消息，可以产生一个flex.messaging.messages.AsyncMessage类的实例，用flex.messaging.MessageBroker发送出去。

示例：客户端可以发布和订阅消息，在messaging-config.xml配置消息目的地，在程序中用“destination”映射关系来访问该类的示例，如示例中的“chat”。注意，通道及其端点(endpoint)真正的定义是在service-config.xml中，messaging-config.xml中仅仅是引用它们。 

可以看到，messaging包括两大类不同的消息通道。分别是streaming channel和polling channel，每一类又有一些扩展类型。

采用polling channel，可以配置为每隔一段时间就轮询一次，也可以配置让服务器在无数据时等待，直到有数据时才响应（长轮询）。轮询响应完成后后就结束连接。

采用streaming channel时，服务器的响应一直保持连接状态，这样服务器一次连接后，可以持续下发数据到客户端。由于HTTP连接不是双工的，所以一条AMF或HTTP通道实际上需要2条浏览器连接，分别用于上行和下行数据。这第二条连接仅在需要向服务器发送数据时才建立，之后立即释放。这种通道特别适合实时性高、客户端刷新的应用。比起轮询，这种模式有效减少了反复建立连接的开销。

IE和Firefox浏览器在每个session最大连接数上有些不同。如果因此导致streaming channel建立失败，则BlazeDS会自动使用messaging-config.xml配置的下一个连接。

实例：定义多个默认通道，作为备份。

如果不在XML中配置消息通道，也可以在Actionscript脚本中动态赋值。

订阅者如何使用selecor属性来制定过滤条件

实现消息机制，可以 AS + Java 也可以 AS + AS

1 AS + Java

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

    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(20);

    } catch (InterruptedException e) {

    }

   }

2 AS + AS 

<mx:Producer id="producer" destination="chat"/>

<mx:Consumer id="consumer" destination="chat" message="messageHandler(event.message)" selector="prop1 = 10"/>

<mx:Script>

<!--[CDATA[

import mx.messaging.messages.AsyncMessage;

import mx.messaging.messages.IMessage;

private function send():void

{

var message:IMessage = new AsyncMessage();

message.headers = new Array();

message.headers["prop1"] = 10;

message.body.chatMessage = msg.text;

producer.send(message);

msg.text = "";

}

private function messageHandler(message:IMessage):void

{

log.text += message.body.chatMessage + "\n";

}

]]-->

</mx:Script>  

WEB-INF\flex\messaging-config.xml 

<destination id="chat"/>

常见错误

[MessagingError message='配置中不存在通道“my-streaming-amf”。']

方案一

报错的代码段

var consumer:Consumer = new Consumer();

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

consumer.subtopic = "tick";

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

consumer.addEventListener(MessageEvent.MESSAGE, messageHandler);

consumer.subscribe();

文件 属性 Flex服务器 应用程序服务器类型 无

改成 J2EE

这个配置在项目根目录 .flexProperties 文件中

缺点：每次发布的时候，都要修改次配置

方案二

var myStreamingAMF:AMFChannel = new StreamingAMFChannel("my-streaming-amf", Constant.portalServerUrl+"/messagebroker/streamingamf");

    var consumer:Consumer = new Consumer();

    consumer.destination = "feed";

    consumer.subtopic = "tick";

    var channelSet:ChannelSet = new ChannelSet(); 

    channelSet.addChannel(myStreamingAMF); 

    consumer.channelSet = channelSet;

    consumer.selector="userName=fortest";

    consumer.addEventListener(MessageEvent.MESSAGE, messageHandler);

    consumer.subscribe();

客户端如何使用选择器

服务端

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

map.put("userName", "fortest");

AsyncMessage msg = new AsyncMessage();

    msg.setDestination("feed");

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

    msg.setHeaders(map);

    msg.setClientId(clientID);

    msg.setMessageId(UUIDUtils.createUUID());

    msg.setTimestamp(System.currentTimeMillis());

    msg.setBody(i+" ");

    msgBroker.routeMessageToService(msg, null);

客户端

var myStreamingAMF:AMFChannel = new StreamingAMFChannel("my-streaming-amf", Constant.portalServerUrl+"/portal/messagebroker/streamingamf");

    var consumer:Consumer = new Consumer();

    consumer.destination = "feed";

    consumer.subtopic = "tick";

    var channelSet:ChannelSet = new ChannelSet(); 

    channelSet.addChannel(myStreamingAMF); 

    consumer.channelSet = channelSet;

    consumer.selector="userName=fortest";

    consumer.addEventListener(MessageEvent.MESSAGE, messageHandler);

    consumer.subscribe();

注意 new StreamingAMFChannel 构造函数中的 portalServerUrl 不要忘记项目域名

报错：Attempt to subscribe or unsubscribe to the subtopic, 'tick', on destination, 'feed', that does not allow subtopics failed.

messaging-config.xml 文件中

<destination id="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>

红色部分代码不能少

报错 [BlazeDS]Endpoint with id 'my-streaming-amf' cannot grant streaming connection to FlexClient with id '9995557A-450B-A6F7-59EF-1FE656A94F1E' because max-streaming-connections-per-session limit of '1' has been reached.

在  service-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="8"/>    

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

    </user-agent-settings>    

   </properties>

  </channel-definition> 

默认是 1 改成 8 即可

报错 打印出来的是 null

private static FeedThread thread;

AsyncMessage msg = new AsyncMessage();

   msg.setDestination("feed");

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

   msg.setClientId(clientID);

   msg.setMessageId(UUIDUtils.createUUID());

   msg.setTimestamp(System.currentTimeMillis());

   msg.setBody("" + i);

   msgBroker.routeMessageToService(msg, null);

   msg.setBody(Object o);//接收的是对象

   private function messageHandler(event:MessageEvent):void      

   {     

    tick = event.message.body as String 但是界面上一般展现的是字符串

//message 成功;     

    txtTick.text = txtTick.text +　tick;      

   }

所以要将整形 i 加一个字符串，否则接界面强制类型转换将打印一堆 null