Blazeds+JMS（ActiveMQ）+Spring实现消息

转载：http://yunzhongxia.iteye.com/blog/565249

 

 

1. 为什么要消息？

工行直连要用到深证通数据交换平台（MDEP），MDEP的请求是不同步的。托管行每天会推过来一些头寸报表信息，这些信息要及时的反应给业务人员。另外，文件扫描和预警都要及时的通知给业务人员。因此，平台利用Blazeds的消息。值得说明的一点是：BlazeDS 是一个基于服务器的 Java 远程控制 (remoting) 和 Web 消息传递 (messaging) 技术，以LGPL（Lesser GNU Public License）公共许可证书发布。它能够使得后端的 Java 应用程序和运行在浏览器上的 Adobe Flex 应用程序相互通信。在Java应用服务器上，它以servlet的形式存在， 因此可以在任何标准Java网络应用中运用它。Blazeds的消息机制是采用不断的轮询来实现的，因此性能不怎么好，并发数最大为100，LCDS的消息机制是采用java的NIO，最高并发数是1000，Blazeds和LCDS相比性能还是比较差的，不过***系统的用户数比较少的情况下，Blazeds还是可以的，毕竟，客户都是比较吝啬的，有免费的肯定不想掏钱买昂贵的LCDS。

 

    2. 消息适配器该用哪一个？

 

<adapters>
        <adapter-definition id="actionscript"
            class="flex.messaging.services.messaging.adapters.ActionScriptAdapter"
            default="true" />
        <adapter-definition id="jms"
            class="flex.messaging.services.messaging.adapters.JMSAdapter" />
    </adapters>

<adapters>
		<adapter-definition id="actionscript"
			class="flex.messaging.services.messaging.adapters.ActionScriptAdapter"
			default="true" />
		<adapter-definition id="jms"
			class="flex.messaging.services.messaging.adapters.JMSAdapter" />
	</adapters>

 

Blazeds提供了两种消息适配器，actionscript和jms.项目一期用的是actionscript适配器，但是有的时候发现文件扫描的消息不能发送到客户端，一直也没有找到原因，关键是Log里面找不到原因。二期应该会有大量的消息要推送到前端，因此打算用jms消息适配器。消息先发送到消息服务器上，flex端订阅消息，如果服务器上有消息在客户端就显示出来。消息服务器很多都是收费的，开源的有ActiveMQ、OPENJMS等。

 

   3.消息服务器该用哪个？

    

　ActiveMQ 是Apache的产品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。在网上查阅了一下关于ActiveMQ 的文档也比较多，而且是开放源代码的，blazeds的官方文档上的例子就是用的ActiveMQ作为消息服务器，因此消息服务器就选它了。

 

   4. 如何搭建环境？

 

      首先要在J2EE服务器（tomcat6）中提供消息服务（JNDI）。在WebRoot/WETA-INFw文件夹下新建一个context.xml.文件的内容如下：

 

<Context privileged="true" antiResourceLocking="false"
    antiJARLocking="false" reloadable="false">
    <!-- Resourced needed for JMS -->
    <Resource name="jms/flex/TopicConnectionFactory"
        type="org.apache.activemq.ActiveMQConnectionFactory"
        description="JMS Connection Factory"
        factory="org.apache.activemq.jndi.JNDIReferenceFactory"
        brokerURL="tcp://localhost:61616"
        brokerName="LocalActiveMQBroker" />
    <Resource name="jms/topic/flex/simpletopic"
        type="org.apache.activemq.command.ActiveMQTopic"
        description="my Topic"
        factory="org.apache.activemq.jndi.JNDIReferenceFactory"
        physicalName="FlexTopic" />
    <Resource name="jms/flex/QueueConnectionFactory"
        type="org.apache.activemq.ActiveMQConnectionFactory"
        description="JMS Connection Factory"
        factory="org.apache.activemq.jndi.JNDIReferenceFactory"
        brokerURL="tcp://localhost:61616"
        brokerName="LocalActiveMQBroker" />
    <Resource name="jms/queue/flex/simplequeue"
        type="org.apache.activemq.command.ActiveMQQueue"
        description="my Queue"
        factory="org.apache.activemq.jndi.JNDIReferenceFactory"
        physicalName="FlexQueue" />
    <!--
    <Valve className="flex.messaging.security.TomcatValve" />
     -->
</Context>

<Context privileged="true" antiResourceLocking="false"
	antiJARLocking="false" reloadable="false">
	<!-- Resourced needed for JMS -->
	<Resource name="jms/flex/TopicConnectionFactory"
		type="org.apache.activemq.ActiveMQConnectionFactory"
		description="JMS Connection Factory"
		factory="org.apache.activemq.jndi.JNDIReferenceFactory"
		brokerURL="tcp://localhost:61616"
		brokerName="LocalActiveMQBroker" />
	<Resource name="jms/topic/flex/simpletopic"
		type="org.apache.activemq.command.ActiveMQTopic"
		description="my Topic"
		factory="org.apache.activemq.jndi.JNDIReferenceFactory"
		physicalName="FlexTopic" />
	<Resource name="jms/flex/QueueConnectionFactory"
		type="org.apache.activemq.ActiveMQConnectionFactory"
		description="JMS Connection Factory"
		factory="org.apache.activemq.jndi.JNDIReferenceFactory"
		brokerURL="tcp://localhost:61616"
		brokerName="LocalActiveMQBroker" />
	<Resource name="jms/queue/flex/simplequeue"
		type="org.apache.activemq.command.ActiveMQQueue"
		description="my Queue"
		factory="org.apache.activemq.jndi.JNDIReferenceFactory"
		physicalName="FlexQueue" />
	<!-- 
	<Valve className="flex.messaging.security.TomcatValve" />
	 -->
</Context>

 

      备注：其实通过上面的配置后，在tomcat6\conf\Catalina\localhost文件下生成了一个utmost.xml文件。关于上面的配置，请继续关注我的博客，我会在以后的写博客的。

 

     修改WEB-INF/flex/messaging-config.xml文件。

<?xml version="1.0" encoding="UTF-8"?>
<service id="message-service" class="flex.messaging.services.MessageService">
    <adapters>
        <adapter-definition id="actionscript"
            class="flex.messaging.services.messaging.adapters.ActionScriptAdapter"
            default="false" />
        <adapter-definition id="jms"
            class="flex.messaging.services.messaging.adapters.JMSAdapter"
            default="true" />
    </adapters>

    <default-channels>
        <channel ref="my-streaming-amf" />
        <channel ref="my-polling-amf" />
    </default-channels>

    <destination id="scanfile" />
    <!-- active MQ stock feed -->
    <destination id="jmsamc"">
        <adapter ref="jms" />
        <properties>
            <!--这里的配置是最关键的，只有durable属性设计为true才能实现持久化订阅-->
            <server>
                <durable>true</durable>
            </server>
            <jms>
                <connection-factory>
                    java:comp/env/jms/flex/TopicConnectionFactory
                </connection-factory>
                <destination-type>Topic</destination-type>
                <destination-jndi-name>
                    java:comp/env/jms/topic/flex/simpletopic
                </destination-jndi-name>
                <message-type>javax.jms.TextMessage</message-type>
                <!-- 持久性 -->
                <delivery-mode>PERSISTENT</delivery-mode>
                <!-- 优先级  -->
                <message-priority>DEFAULT_PRIORITY</message-priority>
                <!--应答模式   -->
                <acknowledge-mode>AUTO_ACKNOWLEDGE</acknowledge-mode>
                <initial-context-environment>
                    <property>
                        <name>Context.SECURITY_PRINCIPAL</name>
                        <value>anonymous</value>
                    </property>
                    <property>
                        <name>Context.SECURITY_CREDENTIALS</name>
                        <value>anonymous</value>
                    </property>
                    <property>
                        <name>Context.INITIAL_CONTEXT_FACTORY</name>
                        <value>
                            org.apache.activemq.jndi.ActiveMQInitialContextFactory
                        </value>
                    </property>
                    <property>
                        <name>Context.PROVIDER_URL</name>
                        <value>tcp://192.168.124.114:61616</value>
                    </property>
                </initial-context-environment>
            </jms>
        </properties>
        <channels>
            <channel ref="my-polling-amf" />
            <channel ref="my-streaming-amf" />
        </channels>
    </destination>
</service>

<?xml version="1.0" encoding="UTF-8"?>
<service id="message-service" class="flex.messaging.services.MessageService">
	<adapters>
		<adapter-definition id="actionscript"
			class="flex.messaging.services.messaging.adapters.ActionScriptAdapter"
			default="false" />
		<adapter-definition id="jms"
			class="flex.messaging.services.messaging.adapters.JMSAdapter"
			default="true" />
	</adapters>

	<default-channels>
		<channel ref="my-streaming-amf" />
		<channel ref="my-polling-amf" />
	</default-channels>

	<destination id="scanfile" />
	<!-- active MQ stock feed -->
	<destination id="jmsamc"">
		<adapter ref="jms" />
		<properties>
			<!--这里的配置是最关键的，只有durable属性设计为true才能实现持久化订阅-->
			<server>
				<durable>true</durable>
			</server>
			<jms>
				<connection-factory>
					java:comp/env/jms/flex/TopicConnectionFactory
				</connection-factory>
				<destination-type>Topic</destination-type>
				<destination-jndi-name>
					java:comp/env/jms/topic/flex/simpletopic
				</destination-jndi-name>
				<message-type>javax.jms.TextMessage</message-type>
				<!-- 持久性 -->
				<delivery-mode>PERSISTENT</delivery-mode>
				<!-- 优先级  -->
				<message-priority>DEFAULT_PRIORITY</message-priority>
				<!--应答模式   -->
				<acknowledge-mode>AUTO_ACKNOWLEDGE</acknowledge-mode>
				<initial-context-environment>
					<property>
						<name>Context.SECURITY_PRINCIPAL</name>
						<value>anonymous</value>
					</property>
					<property>
						<name>Context.SECURITY_CREDENTIALS</name>
						<value>anonymous</value>
					</property>
					<property>
						<name>Context.INITIAL_CONTEXT_FACTORY</name>
						<value>
							org.apache.activemq.jndi.ActiveMQInitialContextFactory
						</value>
					</property>
					<property>
						<name>Context.PROVIDER_URL</name>
						<value>tcp://192.168.124.114:61616</value>
					</property>
				</initial-context-environment>
			</jms>
		</properties>
		<channels>
			<channel ref="my-polling-amf" />
			<channel ref="my-streaming-amf" />
		</channels>
	</destination>
</service>

 

felx前端订阅消息

<mx:ChannelSet id="cs">
        <mx:AMFChannel url="http://192.168.124.114:8099/utmost/messagebroker/amfpolling"/>
        <mx:AMFChannel url="http://192.168.124.114:8099/utmost/messagebroker/streamingamf"/>
    </mx:ChannelSet>

    <mx:Consumer id="consumer" destination="jmsamc" channelSet="{cs}"
         message="messageHandler(event.message)" selector=""/>

<mx:ChannelSet id="cs"> 
        <mx:AMFChannel url="http://192.168.124.114:8099/utmost/messagebroker/amfpolling"/> 
        <mx:AMFChannel url="http://192.168.124.114:8099/utmost/messagebroker/streamingamf"/> 
    </mx:ChannelSet> 
 
    <mx:Consumer id="consumer" destination="jmsamc" channelSet="{cs}" 
         message="messageHandler(event.message)" selector=""/> 
      

 

 

通过以上的配置，基本上实现了Blazeds和JMS的结合。但是还没有解决二期项目的问题，服务器一启动，文件扫描就应该开始；预警是达到一定时间要发送消息给客户端的；ActiveMQ怎么交给Spring管理。下面几篇文章，将会带你慢慢解决这些问题。