- 浏览: 307699 次
- 性别:
- 来自: 北京
文章分类
- 全部博客 (118)
- oracle (16)
- java (13)
- jquery (16)
- hibernate (2)
- spring (14)
- struts (2)
- eclipse (1)
- birt (5)
- integrated skill (13)
- linux (9)
- css+html (5)
- redis (1)
- jqgrid (4)
- mybatis (2)
- multithreading (3)
- jvm (1)
- mvn (3)
- php (2)
- jms (2)
- photoshop (1)
- springmvc (3)
- cluster (1)
- cache (1)
- jquery-ui (2)
- 英语词根 (0)
- svn (0)
- ide (1)
- mysql (3)
- rich-text (1)
- golang (8)
- mac 10 (1)
- android (1)
- react (0)
- bootstrap (2)
- css (2)
- 1 (0)
- beego (1)
- pictures (1)
- java小生转战 (0)
- 转站通知 (0)
- 转站通知,,,啊啊啊 (1)
- threeperson (1)
- unity (1)
最新评论
-
zld406504302:
aaaddddddd
java小生转站threeperson博客 -
zld406504302:
aaaaaddd
java小生转站threeperson博客 -
zld406504302:
aaaaaaaaaaaa
java小生转站threeperson博客 -
乌发舞天:
UEditor 实例 -
van_2013:
请问这是订阅模式的配置吗?
SPRING JMS 整合ACTIVEMQ
近日用spring3.2 jms 与activemq5.8 整合一下,实现了异步发送,异步接收功能,并附上了测试代码
1 )UML 如下
[img]
消息的接受完全是托管到org.springframework.jms.listener.DefaultMessageListenerContainer 中来处理 ,发送client 无需关心消息的接受
[/img]
2 )applicationContext.xml 片段
3)java 相关class 代码
MessageHandler 消息接受的接口
JmsReceiver 消息接受实现类
JmsSender 消息发送 支持异步发送
4) test case
1 )UML 如下
[img]
消息的接受完全是托管到org.springframework.jms.listener.DefaultMessageListenerContainer 中来处理 ,发送client 无需关心消息的接受
[/img]
2 )applicationContext.xml 片段
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"> <!-- 核心线程数,默认为1 --> <property name="corePoolSize" value="5" /> <!-- 最大线程数,默认为Integer.MAX_VALUE --> <property name="maxPoolSize" value="5" /> <!-- 队列最大长度,一般需要设置值>=notifyScheduledMainExecutor.maxNum;默认为Integer.MAX_VALUE --> <property name="queueCapacity" value="1000" /> <!-- 线程池维护线程所允许的空闲时间,默认为60s --> <property name="keepAliveSeconds" value="300" /> <!-- 线程池对拒绝任务(无线程可用)的处理策略,目前只支持AbortPolicy、CallerRunsPolicy;默认为后者 --> <property name="rejectedExecutionHandler"> <!-- AbortPolicy:直接抛出java.util.concurrent.RejectedExecutionException异常 --> <!-- CallerRunsPolicy:主线程直接执行该任务,执行完之后尝试添加下一个任务到线程池中,可以有效降低向线程池内添加任务的速度 --> <!-- DiscardOldestPolicy:抛弃旧的任务、暂不支持;会导致被丢弃的任务无法再次被执行 --> <!-- DiscardPolicy:抛弃当前任务、暂不支持;会导致被丢弃的任务无法再次被执行 --> <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" /> </property> </bean> <!--jms 连接池-- optimizedAckScheduledAckInterval:消息确认周期 --> <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" /> <!--消息发送pojo--> <bean id="messageSender" class="com.cn.ld.modules.jms.worker.JmsSender" /> <!--异步接收的消息监听器--> <bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter"> <constructor-arg> <ref bean="messageReceiver" /> </constructor-arg> </bean>
3)java 相关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 消息接受实现类
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 消息发送 支持异步发送
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"; //开启异步发送 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 jdbcTemplate 批量插入返回自增id
2016-03-22 00:24 1955项目中一个业务场景包含两个相互依赖的批量插入,第二次批量插入 ... -
mvn assembly package jar 无法运行
2015-04-27 20:04 732项目中要跑了一个jar工程,每天做一次数据结 ... -
SPRING INTERCEPTOR 监控 JMS 发送接受性能
2013-07-18 17:17 1550将spring aspectj 和 jms 整合了一下, ... -
SPRING JMS 发送 接受 性能监控
2013-07-18 17:13 0将spring aspectj 和 jms 整合了一下, ... -
SLF4J jar包冲突的迷惑
2013-07-14 12:42 1449Running com.cn.ld.modules.jms.J ... -
SLF4J jar包冲突的迷惑
2013-07-14 12:42 0Running com.cn.ld.modules.jms.J ... -
tomcat 集群下ehcache 整合spring cache-annotation
2013-07-02 12:44 6155系统环境:win7 部署环境:apache2.2 tomcat ... -
springmvc 整合apache tiles
2013-06-15 23:36 9060最近把springmvc3.2.0 和 apache t ... -
springmvc3.2 demo
2013-06-15 23:28 106951:mvn 构建一个web工程 [img] [/img] r ... -
spring jms 异步接受消息
2013-05-22 21:46 24spring jms 的监听器是可配置的,而且通常会交 ... -
SPRING JMS + ACTIVEMQ 实例
2013-05-15 13:21 48工作需要,对spring jms 和消息服务器 amq 做 ... -
Spring Multipart上传
2013-03-18 11:53 11131:在application.xml中初始化CommonsM ... -
@Value 注解
2013-03-13 15:43 1321:在application。xml文件中配置属性文件加载be ... -
spring 事务隔离 与 传播
2012-07-15 21:18 9581) PROPAGATION_REQUIRED ,默认的spr ... -
AXIS2 与SSH 集成 事物代理不可用
2012-05-28 18:42 20462012-5-28 18:17: ... -
spring缓存 ehcache
2012-03-07 10:49 1576ehcache 与spring集成实现简单 ... -
ssh+birt
2011-12-02 12:00 3271新公司工作一年多,接触最多的就是统计,当然用于统计数据展示的b ... -
java,spring,aspectj
2011-11-16 19:39 1146最近要项目有个新需求,统计查询的接口要进行 ...
相关推荐
详细内容: SpringJMS整合ActiveMQ.doc 详细说明文档 apache-activemq-5.8.0-bin.zip ActiveMQ安装包 JMSTest.rar MyEclipse8.5下web工程
spring3.0整合了activemQ和jms可以发送信息和接收消息
前几章我们分别利用spring rmi、httpinvoker、httpclient、webservice技术实现不同服务器间的远程...3) Spring整合activemq配置 4) 定义消息发布者(生产者) 5) 定义消息订阅者(消费者) 6) Spring mvc配置 7) 实例测试
基于Spring+JMS+ActiveMQ+Tomcat,做一个Spring4.1.0和ActiveMQ5.11.1整合实例,实现了Point-To-Point的异步队列消息和PUB/SUB(发布/订阅)模型,简单实例,不包含任何业务。
spring整合JMS-居于ActiveMQ实现
Spring整合JMS(activeMQ)的示例,开发环境为eclipse+maven
Spring整合ActiveMQ超级详细实例,基于Spring+JMS+ActiveMQ+Tomcat,注解的完整实例,包含jar包,可以参考学习
支持持久化的采用Spring整合activeMQ与quartz的JMS数据同步实例,包含依赖的jar包
cJMS之ActiveMQ与Spring整合源码
此实例基于Spring+JMS+ActiveMQ+Tomcat,注解的完整实例,包含jar包
activemq与spring整合发送jms消息入门实例 jar 包和实例代码都在压缩包里了
Spring整合JMS,基于ActiveMQ实现收发消息
SpringBoot2整合Jms超级简单攻略(基于ActiveMQ).。消息主题(Topic),需要一个固定的主题,例如大家都是在谈论外卖,那主题就是外卖,别人讨论其他的跟你无关。 消息队列(MQ) ,存放或者实现JMS的功能,需要用到队列...
java springboot整合activemq工程 #activemq配置 #默认情况下activemq提供的是queue模式 true是可以使用topic,false是仅使用queue模式 spring.jms.pub-sub-domain: true # 设置连接的activemq服务器 spring....
完整jms整合好的spring的java工程,直接运行activeMq就可以运行!
先到官网下载activemq,启动后,就可以直接运行项目文件了
JMS与SPRING的整合实例(基于Apache ActiveMQ) 定义JMS连接工厂 定义JMS Template 定义消息目的地 定义接收监听器 定义一个JMS话题 定义消费者(接收端) 定义发布者 JAVA核心代码一般由三个部分组成: 监听器...
Spring和ActiveMQ的整合实例源码ActiveMQSpringDemo,此实例基于Spring+JMS+ActiveMQ+Tomcat,注解的完整实例,包含jar包
本篇文章主要介绍了详解spring boot整合JMS(ActiveMQ实现),小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧