亚洲免费在线-亚洲免费在线播放-亚洲免费在线观看-亚洲免费在线观看视频-亚洲免费在线看-亚洲免费在线视频

SPRING JMS 整合ACTIVEMQ

系統(tǒng) 2063 0
? 近日用spring3.2 jms 與activemq5.8 整合一下,實(shí)現(xiàn)了異步發(fā)送,異步接收功能,并附上了測(cè)試代碼

1 )UML 如下
?? [img]
??
SPRING JMS 整合ACTIVEMQ
??? 消息的接受完全是托管到org.springframework.jms.listener.DefaultMessageListenerContainer 中來處理?? ,發(fā)送client 無需關(guān)心消息的接受
?? [/img]
2 )applicationContext.xml 片段
    
 <bean id="taskExecutor"
		class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
		<!-- 核心線程數(shù),默認(rèn)為1 -->
		<property name="corePoolSize" value="5" />
		<!-- 最大線程數(shù),默認(rèn)為Integer.MAX_VALUE -->
		<property name="maxPoolSize" value="5" />
		<!-- 隊(duì)列最大長(zhǎng)度,一般需要設(shè)置值>=notifyScheduledMainExecutor.maxNum;默認(rèn)為Integer.MAX_VALUE -->
		<property name="queueCapacity" value="1000" />
		<!-- 線程池維護(hù)線程所允許的空閑時(shí)間,默認(rèn)為60s -->
		<property name="keepAliveSeconds" value="300" />
		<!-- 線程池對(duì)拒絕任務(wù)(無線程可用)的處理策略,目前只支持AbortPolicy、CallerRunsPolicy;默認(rèn)為后者 -->
		<property name="rejectedExecutionHandler">
			<!-- AbortPolicy:直接拋出java.util.concurrent.RejectedExecutionException異常 -->
			<!-- CallerRunsPolicy:主線程直接執(zhí)行該任務(wù),執(zhí)行完之后嘗試添加下一個(gè)任務(wù)到線程池中,可以有效降低向線程池內(nèi)添加任務(wù)的速度 -->
			<!-- DiscardOldestPolicy:拋棄舊的任務(wù)、暫不支持;會(huì)導(dǎo)致被丟棄的任務(wù)無法再次被執(zhí)行 -->
			<!-- DiscardPolicy:拋棄當(dāng)前任務(wù)、暫不支持;會(huì)導(dǎo)致被丟棄的任務(wù)無法再次被執(zhí)行 -->
			<bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />
		</property>
	</bean>
          <!--jms 連接池--
            optimizedAckScheduledAckInterval:消息確認(rèn)周期

           -->
          
	<bean id="jmsConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
		<property name="connectionFactory">
			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
				<property name="brokerURL" value="tcp://localhost:61616" />
				<property name="closeTimeout" value="60000" />
				<property name="userName" value="admin" />
				<property name="password" value="admin" />
				<!--<property name="optimizeAcknowledge" value="true" />-->
				<property name="optimizedAckScheduledAckInterval" value="10000" />
			</bean>
		</property>
	</bean>

	<!-- Spring JMS Template -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory">
			<ref local="jmsConnectionFactory" />
		</property>
	</bean>
         
	         <!--queue通道-->
	<bean id="asyncQueue" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg index="0">
			<value>asyncQueue</value>
		</constructor-arg>
	</bean>
         <!--topic通道-->
	<bean id="asyncTopic" name="asyncTopic"
		class="org.apache.activemq.command.ActiveMQTopic">
		<constructor-arg index="0">
			<value>asyncTopic</value>
		</constructor-arg>
	</bean>
         <!--消息接受容器,多線程異步接受消息-->
	<bean id="jmsContainer"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="jmsConnectionFactory" />
		<property name="destination" ref="asyncTopic" />
		<property name="messageListener" ref="messageListener" />
		<property name="sessionTransacted" value="false" />
	</bean>
         <!--消息接受pojo-->
	<bean id="messageReceiver" class="com.cn.ld.modules.jms.worker.JmsReceiver" />
         <!--消息發(fā)送pojo-->
	<bean id="messageSender" class="com.cn.ld.modules.jms.worker.JmsSender" />
          <!--異步接收的消息監(jiān)聽器-->
	<bean id="messageListener"
		class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<constructor-arg>
			<ref bean="messageReceiver" />
		</constructor-arg>
	</bean>

  



3)java 相關(guān)class 代碼
??? MessageHandler 消息接受的接口
?
    
package com.cn.ld.modules.jms.handler;

import java.io.Serializable;

public interface MessageHandler {
	void receive(TextMessage message);
	
	
	void handleMessage(String message);

	void handleMessage(Map<String, Object> message);

	void handleMessage(byte[] message);

	void handleMessage(Serializable message);
}

  
  


? JmsReceiver? 消息接受實(shí)現(xiàn)類
?
    

   package com.cn.ld.modules.jms.worker;

import java.io.Serializable;

public class JmsReceiver implements MessageHandler {
	private Collection<String> collection;

	@Override
	public void receive(TextMessage message) {
		try {
			if (collection == null) {
				this.collection = new ArrayList<String>();
			}
			collection.add(message.getText());
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
	
	
	@Override
	public void handleMessage(String message) {
		/*
		 * if(collection == null){ this.collection = new ArrayList<String>(); }
		 * collection.add(message);
		 */
	}

	@Override
	public void handleMessage(Map<String, Object> message) {
		Set<String> keySet = message.keySet();
		Iterator<String> keys = keySet.iterator();
		while (keys.hasNext()) {
			String key = keys.next();
			System.out.println(message.get(key));
		}

	}

	@Override
	public void handleMessage(byte[] message) {

	}

	@Override
	public void handleMessage(Serializable message) {
	}

	public Collection<String> getCollection() {
		return collection;
	}

	public void setCollection(Collection<String> collection) {
		this.collection = collection;
	}

}

  
  

??
? JmsSender 消息發(fā)送 支持異步發(fā)送
?
    
package com.cn.ld.modules.jms.worker;

import java.util.Collection;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.util.Assert;

import com.cn.ld.modules.annotation.MethodMonitorCount;

public class JmsSender {

	@Autowired
	private JmsTemplate jmsTemplate;
        
	@Autowired
	private TaskExecutor taskExecutor;

	private Destination destination;

	private boolean isSendAsync = false;

	public JmsSender(){}
	public JmsSender(Destination destination) {
		if (null == destination)
			this.destination = new ActiveMQTopic("topic");
		else
			this.destination = destination;
	}

	
	public void sendSingle(String message,Destination destination) {
		sendMessage(message,destination);
	}

	public void sendBatch(Collection<?> messages,Destination destination) {
		Assert.notNull(messages, "param 'messages' can't be null !");
		Assert.notEmpty(messages, "param 'message' can't be empty !");
		for (Object message : messages) {
			if (null != message && message instanceof String) {
				sendSingle(String.valueOf(message),destination);
			}
		}
	}

	private void sendMessage(final String message,Destination destination) {
		final Destination sendDest = destination ;
		if (isSendAsync) {
			taskExecutor.execute(new Runnable() {
				@Override
				public void run() {
					send(message,sendDest);
				}
			});
		} else {
			send(message,destination);
		}
	}

	private void send(final String message,Destination destination) {
		this.jmsTemplate.send(destination, new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(message);
			}

		});
	}

	public boolean isSendAsync() {
		return isSendAsync;
	}
	
	public void setSendAsync(boolean isSendAsync) {
		this.isSendAsync = isSendAsync;
	}

	public Destination getDestination() {
		return destination;
	}

}

  
  

??

4) test case
    
package com.cn.ld.modules.jms;

import java.io.File;
import java.io.IOException;

import org.apache.activemq.command.ActiveMQTopic;
import org.apache.log4j.Logger;
import org.aspectj.util.FileUtil;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import com.cn.ld.modules.jms.worker.JmsSender;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "classpath:applicationContext.xml" })
public class JmsTest {
	protected final Logger logger = Logger.getLogger(this.getClass());

	@Autowired
	private JmsSender jmsSender;

	private String destination;
	private int no = 10* 10000;
	private String message;

	@Before
	public void init() throws IOException {
		String filePath = Thread.currentThread().getContextClassLoader()
				.getResource("").getPath()
				+ "message.txt";
		message = FileUtil.readAsString(new File(filePath));
		this.destination = "asyncTopic";
                  //開啟異步發(fā)送
		this.jmsSender.setSendAsync(true);
	}

	@Test
	public void send() throws InterruptedException {
		ActiveMQTopic dest = new ActiveMQTopic(this.destination);
		for (int i = 0; i < no; i++) {
			jmsSender.sendSingle(message, dest);
		}
		Thread.sleep(1000000000);
	}

}


  

?

SPRING JMS 整合ACTIVEMQ


更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主

微信掃碼或搜索:z360901061

微信掃一掃加我為好友

QQ號(hào)聯(lián)系: 360901061

您的支持是博主寫作最大的動(dòng)力,如果您喜歡我的文章,感覺我的文章對(duì)您有幫助,請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧,狠狠點(diǎn)擊下面給點(diǎn)支持吧,站長(zhǎng)非常感激您!手機(jī)微信長(zhǎng)按不能支付解決辦法:請(qǐng)將微信支付二維碼保存到相冊(cè),切換到微信,然后點(diǎn)擊微信右上角掃一掃功能,選擇支付二維碼完成支付。

【本文對(duì)您有幫助就好】

您的支持是博主寫作最大的動(dòng)力,如果您喜歡我的文章,感覺我的文章對(duì)您有幫助,請(qǐng)用微信掃描上面二維碼支持博主2元、5元、10元、自定義金額等您想捐的金額吧,站長(zhǎng)會(huì)非常 感謝您的哦!!!

發(fā)表我的評(píng)論
最新評(píng)論 總共0條評(píng)論
主站蜘蛛池模板: 六月丁香婷婷激情国产 | 久久www免费人成_看片高清 | 久久99爱re热视 | 欧洲色网| 国产成人a∨麻豆精品 | 夜夜摸天天操 | 国产精品国产亚洲精品看不卡 | 在线观看日本免费视频大片一区 | 亚洲精品一区二区久久这里 | 久久一区不卡中文字幕 | 国产一及毛片 | 一级特级欧美aaaaa毛片 | 精品精品国产自在久久高清 | 久久精品久久精品久久精品 | 九草伊人 | a级亚洲片精品久久久久久久 | 在线观看视频中文字幕 | 日日干天天草 | 久久国产精品99久久久久久牛牛 | 99热播| 免费看在线爱爱小视频 | 婷婷久月 | 亚洲精品欧洲精品 | 中文字幕中文字幕中中文 | 天天久久综合 | 国产精品成aⅴ人片在线观看 | 国产亚洲精品一区二区在线播放 | 一级在线免费视频 | 人人澡人人干 | 日本综合色 | 国产一级影片 | 久久美剧| 亚洲精品视频一区二区 | a在线观看免费视频 | 4虎最新网站 | 欧美综合专区 | 色爱综合网欧美 | 高清中文字幕免费观在线 | 欧美又黄又嫩大片a级 | 婷婷爱五月| 国产精品短视频 |