ActiveMQ spring 配置

<bean id="memcachedClient" class="net.spy.memcached.spring.MemcachedClientFactoryBean" >    
<property name="servers" value="memcache.mtc.sohu.com:33001"/>  
<property name="transcoder">    
<bean class="net.spy.memcached.transcoders.SerializingTranscoder">        
<property name="compressionThreshold" value="1024"/>      
</bean>   
</property>  
<property name="opTimeout" value="3000"/>    
<property name="timeoutExceptionThreshold" value="1998"/>   
<property name="hashAlg" value="KETAMA_HASH"/>    
<property name="locatorType" value="CONSISTENT"/>     
<property name="failureMode" value="Redistribute"/>    
<property name="useNagleAlgorithm" value="false"/>  
    </bean>
    <!-- static service util -->
    <bean id="memcache" class="com.sohu.t.api.memcache.MemcacheUtil" factory-method="getInstance" />
    <!-- jms change password 相关 -->
    <!-- 配置JMS连接工厂 -->
    <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
         <property name="brokerURL" value="tcp://ip:port"/>
         <property name="userName" value="rcver"/>
         <property name="password" value="r%@#$cv_09ers"/>
         <property name="clientID" value="1073-PASSPORT.PASSWORD.TOPI"/>
    </bean>
    <!-- 配置JMS模版 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
         <property name="connectionFactory" ref="connectionFactory"/>
         <property name="defaultDestination" ref="destination"></property>
    </bean>
    <bean id="topicListener"
class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicMessageReceiver" />
<property name="defaultListenerMethod" value="receive" />
</bean>
    <!-- 发送消息的目的地（一个队列） -->
    <bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">
         <!-- 设置消息队列的名字 -->
<!--         <constructor-arg index="0" value="PASSPORT.PASSWORD.TOPIC"/> -->
<!--         <property name="physicalName" value="PASSPORT.PASSWORD.TOPIC?consumer.retroactive=true&amp;consumer.prefetchSize=10"></property> -->
          <property name="physicalName" value="PASSPORT.PASSWORD.TOPIC"></property>
    </bean>
    <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">   
        <property name="connectionFactory" ref="connectionFactory" />   
        <property name="pubSubDomain" value="true"/>   
        <property name="destination" ref="destination" />   
        <property name="clientId" value="1073-PASSPORT.PASSWORD.TOPI"/> 
        <property name="messageListener" ref="topicListener" />
        <property name="sessionAcknowledgeModeName" value="AUTO_ACKNOWLEDGE"/>
        <property name="subscriptionDurable" value="true"/> 
        <property name="durableSubscriptionName" value="PASSPORT.PASSWORD.TOPIC" /> 
   </bean>
   <bean id="topicMessageReceiver" class="com.sohu.t.api.activemq.MessageReceiver">
   <constructor-arg ref="userSecurityTaskExecutor" />
   </bean>



import javax.jms.Destination;
import javax.jms.TextMessage;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;

import com.sohu.t.api.push.mq.ActivityTaskExecutor;
import com.sohu.t.api.push.mq.MsgTaskExecutor;
import com.sohu.t.api.push.mq.UserSecurityTaskExecutor;
import com.sohu.t.api.push.queue.KestrelQueueFactory;
import net.rubyeye.xmemcached.MemcachedClient;
import com.sohu.t.api.push.util.Constants;


public class Server extends Thread {

private MsgTaskExecutor msgExecutor;
private ActivityTaskExecutor activityExecutor;
private UserSecurityTaskExecutor userSecurityTaskExecutor;

JmsTemplate template;
Destination destination;

private MemcachedClient memcachedClient = KestrelQueueFactory.getInstance();
private static final ApplicationContext context = new ClassPathXmlApplicationContext(new String[] {"applicationContext.xml"});
public Server(){
msgExecutor = (MsgTaskExecutor)context.getBean("msgTaskExecutor");
activityExecutor = (ActivityTaskExecutor)context.getBean("activityTaskExecutor");
userSecurityTaskExecutor = (UserSecurityTaskExecutor)context.getBean("userSecurityTaskExecutor");

//用户passport密码修改业务
template = (JmsTemplate) context.getBean("jmsTemplate");
        destination = (Destination) context.getBean("destination");
}
@Override
    public void run() {
while(true){
try{
// String jsonStr = memcachedClient.get(Constants.QUEUE_NAME);
// if(jsonStr!=null){
// msgExecutor.excuteMsgExecutor(jsonStr);
// }
//
TextMessage changPasswordTM = (TextMessage) template.receive(destination);
                if (null != changPasswordTM) {
                String changPasswordMsg = changPasswordTM.getText();
                System.out.println("收到消息内容为: " + changPasswordMsg);
                userSecurityTaskExecutor.excuteMsgExecutor(changPasswordMsg);
                 }
// if(jsonStr==null){
// Thread.sleep(1000);
// }
}catch(Exception e){
e.printStackTrace();
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
e1.printStackTrace();
continue;
}
continue;
}
}
    }
public static void main(String args[]){
Server thread = new Server();
thread.start();
}
}