? 近日用spring3.2 jms 與activemq5.8 整合一下,實(shí)現(xiàn)了異步發(fā)送,異步接收功能,并附上了測(cè)試代碼
1 )UML 如下
?? [img]
??
??? 消息的接受完全是托管到org.springframework.jms.listener.DefaultMessageListenerContainer 中來處理?? ,發(fā)送client 無需關(guān)心消息的接受
?? [/img]
2 )applicationContext.xml 片段
3)java 相關(guān)class 代碼
??? MessageHandler 消息接受的接口
?
? JmsReceiver? 消息接受實(shí)現(xiàn)類
?
??
? JmsSender 消息發(fā)送 支持異步發(fā)送
?
??
4) test case
?
1 )UML 如下
?? [img]
??

??? 消息的接受完全是托管到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); } }
?
更多文章、技術(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ì)您有幫助就好】元
