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

参考文章

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);//接收的是对象 但是界面上一般展现的是字符串

//message 成功
   private function messageHandler(event:MessageEvent):void      
   {     
    tick = event.message.body as String;     
    txtTick.text = txtTick.text +　tick;      
   }

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