spring+ActiveMQ+JMS+线程池实现简单的分布式,多线程,多任务的异步任务处理系统
2016-08-26 11:11
771 查看
前言:随着系统的业务功能不断增强,传统的单机、单任务,单线程的运行模式已经逐渐的被淘汰,取而代之的是分布式,多任务,多线程,当然,现在开源的这方面的框架也非常的多,大概的思想也都类似,下面就结合我这一年多的工作心得,分享一个简单易实现的分布式,多任务,多线程的异步任务处理系统的基本实现。
1.系统部署图
该系统主要由3部分构成,任务生产者集群,消息中间件集群,任务消费者集群,下面来分别说下这3部分的作用:
任务生产者集群:顾名思义,主要用来产生消息任务,并将这些消息任务发送到消息中间件集群中,任务生产者集群可能使用的是不同的开发语言,开发框架以及不同的开发平台。
消息中间件集群:消息中间件集群主要的作用是使生产者和消费者解耦,屏蔽各个异构系统之间的区别,以及保证消息的传送,超时重试和消息任务的负载均衡。并确保同一消息只被一个消费者消费。为了实现系统的高可用性,此处使用了集群模式,实现master-slave模式。
任务消费者集群:这个是我们的业务核心,通过消费消息中间件传送过来的消息,从而实现我们的业务功能需求。为了保证任务被及时的处理,我们会用到spring的线程池,来实现任务的异步调度。
2.系统设计
(1)为了保证系统的高可用性,消息中间件集群我们采用的是主备结构,配置文件如下:
<!-- 通过往PooledConnectionFactory注入一个ActiveMQConnectionFactory可以用来将Connection,Session和MessageProducer池化
这样可以大大减少我们的资源消耗, -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory" />
<property name="maxConnections" value="20" />
</bean>(2),为了及时的监听消息,我们使用到了JMS中的MessageListener,当然,为了有更好的扩展性和灵活性,我们可以使用SessionAwareMessageListener以及MessageListenerAdapter来实现消息驱动POJO,示例代码如下:
对于接收到的消息会采用异步的任务调度结合线程池来处理,示例代码如下:
(4)消息确认机制
为了保证消息或者是请求被至少处理一次,可以引入消息的确认机制,JMS总共为我们提供了3种确认机制,分别如下:
Auto_acknowledge:JMS客户端会自动向服务器发送确认消息,如果服务器没有接收到这个确认消息,就会认为该消息未被传送,并可能会试图重新发送。
Client_acknowledge:Auto_acknowledge模式中,确认总是隐式的在onMessage处理器返回之后发生,而Client_acknowledge则是由客户端控制何时发送确认,这样的话,可以保证接收消息的客户端能够实现对“保证消息传送”更细粒度的控制。当然,这种方式需要客户端来显示的发送,例如调用message.acknowledge();方法
Dups_OK_acknowledge:如果在会话上指定这种模式的话,JMS提供者可以将一条消息向统一目的地发送两次以上,这与前面两种模式的“一次且仅仅一次”的语义就不同了,用于可以接收重复消息的程序。
(5)消息策略
由于消息生产者可能生产的消息各不一样,例如TextMessage、MapMessage等,可以根据不同的消息使用不同的策略,示例代码如下:
由于消息消费者是以集群模式在运行,那么具体到每一条消息,该有哪台机器来消费了,这个就涉及到消息的负载均衡,在系统中,可以利用JMS提供者从消息源头上来实现,具体的负载均衡算法会因JMS提供商的不同而不同,但大概主流的几种算法如下:哈希散列算法,轮询调度算法,first-available均衡算法,使用的时候,需要查阅JMS提供商的文档来确定。
通过上面的这几步,就基本上实现了一个简单的分布式,多任务,多线程的异步任务处理系统,整体运行结果如下:
1.系统部署图
该系统主要由3部分构成,任务生产者集群,消息中间件集群,任务消费者集群,下面来分别说下这3部分的作用:
任务生产者集群:顾名思义,主要用来产生消息任务,并将这些消息任务发送到消息中间件集群中,任务生产者集群可能使用的是不同的开发语言,开发框架以及不同的开发平台。
消息中间件集群:消息中间件集群主要的作用是使生产者和消费者解耦,屏蔽各个异构系统之间的区别,以及保证消息的传送,超时重试和消息任务的负载均衡。并确保同一消息只被一个消费者消费。为了实现系统的高可用性,此处使用了集群模式,实现master-slave模式。
任务消费者集群:这个是我们的业务核心,通过消费消息中间件传送过来的消息,从而实现我们的业务功能需求。为了保证任务被及时的处理,我们会用到spring的线程池,来实现任务的异步调度。
2.系统设计
(1)为了保证系统的高可用性,消息中间件集群我们采用的是主备结构,配置文件如下:
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="failover:(tcp://localhost:61616,tcp://localhost:61617)" /> </bean>为了减少消息中间件由于频繁连接导致的性能消耗,会使用连接池,配置文件如下:
<!-- 通过往PooledConnectionFactory注入一个ActiveMQConnectionFactory可以用来将Connection,Session和MessageProducer池化
这样可以大大减少我们的资源消耗, -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory" />
<property name="maxConnections" value="20" />
</bean>(2),为了及时的监听消息,我们使用到了JMS中的MessageListener,当然,为了有更好的扩展性和灵活性,我们可以使用SessionAwareMessageListener以及MessageListenerAdapter来实现消息驱动POJO,示例代码如下:
public class ConsumerSessionAwareMessageListener implements SessionAwareMessageListener<TextMessage> { private Destination destination; @Override public void onMessage(TextMessage message, Session session) throws JMSException { try { String receiveMessage = ((TextMessage) message).getText(); // 创建消息生产者,用来发送回复消息到回复队列里面 MessageProducer producer = session.createProducer(destination); producer.send(session.createTextMessage("消费者回复消息!")); System.out.println("消费者收到的消息为:"+receiveMessage); } catch (JMSException e) { e.printStackTrace(); } } }(3)异步任务调度
对于接收到的消息会采用异步的任务调度结合线程池来处理,示例代码如下:
@SuppressWarnings({"unchecked", "rawtypes"}) public class ConsumerReceive implements MessageListener { private CustomerServiceStrategyI strategy; @Override @Async("mqExecutor")// 异步的任务处理 public void onMessage(Message message) { System.out.println("当前处理任务的线程为:" + Thread.currentThread().getName()); if (message instanceof TextMessage) { strategy = new TextMessageStrategy(); strategy.doService(message); message.acknowledge();// 客户端消息确认机制 } } }异步任务线程池的配置如下:
<task:annotation-driven/> <task:executor id="mqExecutor" pool-size="5-10" queue-capacity="20000" keep-alive="2000" rejection-policy="CALLER_RUNS"/>注意:@Async("mqExecutor")这个注解表示该方法会通过异步的方式来执行,会直接跳过主程序
(4)消息确认机制
为了保证消息或者是请求被至少处理一次,可以引入消息的确认机制,JMS总共为我们提供了3种确认机制,分别如下:
Auto_acknowledge:JMS客户端会自动向服务器发送确认消息,如果服务器没有接收到这个确认消息,就会认为该消息未被传送,并可能会试图重新发送。
Client_acknowledge:Auto_acknowledge模式中,确认总是隐式的在onMessage处理器返回之后发生,而Client_acknowledge则是由客户端控制何时发送确认,这样的话,可以保证接收消息的客户端能够实现对“保证消息传送”更细粒度的控制。当然,这种方式需要客户端来显示的发送,例如调用message.acknowledge();方法
Dups_OK_acknowledge:如果在会话上指定这种模式的话,JMS提供者可以将一条消息向统一目的地发送两次以上,这与前面两种模式的“一次且仅仅一次”的语义就不同了,用于可以接收重复消息的程序。
(5)消息策略
由于消息生产者可能生产的消息各不一样,例如TextMessage、MapMessage等,可以根据不同的消息使用不同的策略,示例代码如下:
策略接口:为了更好的兼容性,此处使用了泛型 package com.chhliu.myself.activemq.start.async; public interface CustomerServiceStrategyI<P, V> { P doService(V message); } 具体的策略类: package com.chhliu.myself.activemq.start.async; import javax.jms.JMSException; import javax.jms.TextMessage; public class TextMessageStrategy implements CustomerServiceStrategyI<User, TextMessage> { @Override public User doService(TextMessage message) { try { String receiveMessage = message.getText(); System.out.println("消费者收到的消息为:"+receiveMessage); return null; } catch (JMSException e) { } return null; } }(6)负载均衡
由于消息消费者是以集群模式在运行,那么具体到每一条消息,该有哪台机器来消费了,这个就涉及到消息的负载均衡,在系统中,可以利用JMS提供者从消息源头上来实现,具体的负载均衡算法会因JMS提供商的不同而不同,但大概主流的几种算法如下:哈希散列算法,轮询调度算法,first-available均衡算法,使用的时候,需要查阅JMS提供商的文档来确定。
通过上面的这几步,就基本上实现了一个简单的分布式,多任务,多线程的异步任务处理系统,整体运行结果如下:
================生产者创建了一条消息============== ================生产者创建了一条消息============== 当前处理任务的线程为:mqExecutor-2 消费者收到的消息为:hello acticeMQ:my name is chhliu!fcb25e99-1181-48bb-963f-8d20a98829ab 当前处理任务的线程为:mqExecutor-1 消费者收到的消息为:hello acticeMQ:my name is chhliu!12578be1-56a4-4b41-a8d4-112031617525 ================生产者创建了一条消息============== ================生产者创建了一条消息============== 当前处理任务的线程为:mqExecutor-5 消费者收到的消息为:hello acticeMQ:my name is chhliu!24d8f8c2-b172-4f7e-81e8-d87b73b7825b 当前处理任务的线程为:mqExecutor-3 消费者收到的消息为:hello acticeMQ:my name is chhliu!00dc0934-c521-42da-83fb-9f541f667ec9其实上面的这个简单的系统,也可以当成简单的RPC来使用,如果要实现更细粒度的RPC的话,可以引进Netty等来实现,当然这些都是后话了。
相关文章推荐
- 解密Java内存溢出之持久代
- jdk、jre和jvm
- Java定时任务的三种实现方法
- 一个简单的基于注解的 Controller (spring 2.5 annotation 方式实现mvc )
- C#与C#接口对接,C#与Java的接口对接(序列化与反序列化)——(二)
- HashMap的工作原理jdk1.8-源码调试
- java web --fileupload插件网页文件管理
- Java JVM:内存溢出(栈溢出,堆溢出,持久代溢出以及 nable to create native thread)
- java Robot 类模拟键盘按键和鼠标点击(全局模拟)
- springmvc的简单登录功能
- java集合-HashTable
- C#与C#接口对接,C#与Java的接口对接(序列化与反序列化)——(一)
- spring data jpa使用详解
- java web 文件上传和下载
- Spring MVC 学习笔记(二) 基于spring2.5注解实现的spring MVC项目
- Spring MVC 学习笔记(一) 基于spring2.5的纯xml配置
- 有关Java学习的记录Two
- HashMap的工作原理JDK1.6
- springMVC + velocity 的结合
- Spring JDBCTemplate学习笔记