java基于线程的分布式

 

java基于线程的分布式

1.      引言

1.1. 背景

有的任务比较消耗资源，需要将任务分散在不同的机器上运行，充分利用硬件资源。

例如下载任务、例如计算1---1万亿的和。

2.      总体思路

任务分发服务器1个、工作客户端若干个

见图：

 

 

 

2.1. 任务分发服务器

负责任务的分发、维护各个客户端的状态。

将应用程序中的任务，添加到队列中，通过策略分发任务给工作线程，维护客户端状态、维护任务状态，对超时的任务等进行处理。

2.2. 工作客户端

在启动的时候向任务分发服务器注册，并开启一定数量的线程池，等待任务分发服务器分配任务。

根据配置文件里的配置，想服务器注册，提供任务执行线程，保持与服务器的会话，维护与服务器的状态，提供自身硬件资源的说明，方便服务器端制定策略分发任务。

3.      详细设计

3.1. 传送对象

3.1.1.    传送任务对象

利用java的对象流传送对象，将服务器端的任务以对象的方式传送给客户端，客户端同样适用对象流接收任务，并复原对象的各个属性。

ObjectOutputStream和ObjectInputStream

ObjectOutputStream在传送对象的时候要求对象是实现了序列化接口的，这要求在编写任务的时候需要实现此接口。

ObjectInputStream在接收对象的时候，需要能够找到对应的类定义，这需要客户端启动之后，能够动态的加载任务类，可以考虑自定义Classloader，在特定的目录中加载任务类定义。这个过程也是由服务器端发起，服务器端在发送任务对象前，先判断任务对象的类定义（字节码）是否已经发送给了客户端，如果没有，则先发送字节码定义给客户端。

3.1.2.    传送工作相关对象

在使用ObjectInputStream，接收对象的时候，不仅需要对象本身能够被加载，对象中使用到的其他类也必须能够被加载。

如：传送对象a，他的类型A，在a中调用了b的方法，b的类型是B，那么ObjectInputStream接收a对象的时候，必须能够加载A、B两个类，才能接受正常。

那么这就需要能够在传送a之前，将a对象嵌套使用的类定义全部传送，这一点暂时没有想到好的办法。（要是有好的建议，请说明。。。）

3.2. 自定义classloader

使用对象流传送对象，在客户端读取对象的时候，需要对内存中能够加载传送过来的任务类，如果在客户端无没有定义传送的任务类，则会跑出ClassCastException，在客户端，采用自定义的Classloader，目的是能够从临时目录中加载传送的任务类定义，在服务器端端，每次分发任务前，会先判断传送的任务类是否已经传送到这个客户端，如果没有，则会首先将任务类的字节码传送到客户端，客户端在接受到任务类的字节码的时候，会将字节码放入到客户端的临时目录里，在后续的接受任务对象的时候，才能够从临时目录中加载对应的类定义，从而实例化并复原从服务器端发送的任务对象。

3.3. 任务失败拦截

在客户端执行的任务，如果在执行的过程中出现异常，会自动向服务器端发送异常报告，说明是哪一个任务没有能够正常的执行完成，将错误信息发送到服务器端，服务器接收到这种异常报告，会将这个任务再次分配，将任务再一次的交给被的客户端去执行。当然，如果任务正常执行结束，客户端也会自动向服务器报告状态，说明任务正常完成，服务器在接收到这种报告的时候，会清除对应的任务再服务器端的状态等信息，这就是整个任务的生命周期。

 

3.4. 任务的生命周期

1、  初始化的任务，会添加到服务器的队列中，等待服务器分发任务

2、  服务器分发任务到一台客户端上，并且表示这个任务的状态

3、  客户端执行任务，并返回任务的状态

4、  如果客户端没有完成任务，会将任务重新交回到服务器端，服务器会执行步骤2

5、  如果客户端正常完成任务，服务器会在接收到状态报告之后，清除任务的状态标识，任务结束

 

 

4.      使用

4.1. 服务器端

4.1.1.    导入依赖包

apache-commons-discovery.jar

apache-commons-id.jar

commons-logging-1.0.4.jar

log4j-1.2.15.jar

 

distribution.jar

4.1.2.    编写任务类  

 例子：

public class CalculateWork extends DistributionSupport implements Serializable {

	private static final long serialVersionUID = 1L;
	
	@Override
	public void run() {
		/*
		 * 在run的这一层，不处理异常
		 * 当异常发生的时候，客户端会想服务器报告错误，服务器会重新分发任务
		 */
		
		int maxNum = new Random().nextInt(100000);
		int sum = 0;
		for(int i = 1; i < maxNum; i++){
			sum += i;
		}
		String message = "计算结果:1---" + maxNum + "的总和为" + sum + "";
		
		
		if(new Random().nextInt(100) < 30){//概率任务失败
			@SuppressWarnings("unused")
			int errorInt = 1 / 0;
		}else{
			//封装的向服务器传送对象的方法
			//向服务器端发送一个打印请求
			eventToServer(new Event(EventType.W_ECHO_MESSAGE, message, this, null));
		}
	}

}

 

4.2. 工作客户端

4.2.1.    配置客户端

配置distribution.properties文件

serverIp=127.0.0.1//服务器ip地址 
serverPort=10004//服务器监听端口
clientThreadPoolCount=10//客户端工作的线程池大小  

 

 

4.2.2.    启动客户端

distribution.jar既是服务器端，同时也是一个可运行jar包，直接运行就可以

jar –jar distribution.jar

当然运行的时候需要保证目录中有distribution.properties文件和distribution_lib文件夹（依赖性jar包）

 

5.      运行截图

5.1. 启动客户端

 

  

5.2. 启动服务器端

 

 

5.3. 服务器分发3个任务

 

 

5.4. 客户端处理任务

 

 

6.      目前版本说明：

1、  没有实现任务相关对象的传递，即在任务中，不能使用别的类中的方法，也不能别的类的定义。

2、  服务器端和工作客户端的通信是阻塞式IO

3、  任务的分发是轮询机制，没有加入工作客户端的cpu、内存等资源的参考，以后可以考虑在分发机制上根据工作客户端的状态进行分发

 

 

源码、测试程序见distribuition.rar

 

 

 

评论

21 楼 城的灯 2011-03-16  

采用mina直接传递对象，使用NIO比楼主自己的序列化要好些，我个人觉得。同样在解决多个对象的依赖上面还需要寻找更好的方法！

20 楼 月色正浓 2011-03-16  

mxswl 写道

1.强制保证客户端与服务器处于相同的基础ClassPath环境下，可以通过时间检查点来保证这一点。

关于时间检查点可以说详细点吗?

19 楼 jimichan 2011-03-14  

hadoop

mapreduce

存储+运算

18 楼 sunheavenvan 2011-03-14  

cjnetwork 写道

sunheavenvan 写道

个人感觉把一个Thread看作Task，只是把Task粒度放大了而已，不过真的，与Task有何区别？若果你需要发送有序的操作集合，干嘛不用队列呢？队列=有序集合，如发送的是TaskQueue如何？

当然不一定就必须使用线程作为任务的载体，也可以使用Task，或是别的（如设计一个接口，在客户端复原对象的时候，强制转换被传输对象为接口，然后调用接口方法）。不过个人感觉使用线程还是有一定的优势的，这种分布式本来就是处理比较耗时的任务（或是叫做工作），当这样的工作发送给客户端执行的时候，客户端再接收到这样的任务的时，一定不能以阻塞式的执行这样一个任务，然后返回结果给服务器端，因为服务器端可能源源不断在传输这样的多个任务到客户端，并且这些任务不具有任何联系，各自完成一部分逻辑操作。客户端再接收到这样的工作后，需要单独开启线程资源来执行这些任务，并行的。

使用队列这一点，你能讲的具体一点么？我这里没有怎么理解，使用队列能满足这样的需求么，如何实现？
假如有这样三个事情需要处理：
1、计算1---1000的累加和，保存结果到本地磁盘文件中d:/calculateResult.txt
2、下载http://www.baidu.cn网页内容,保存结果到磁盘文件d:/downloadResult.txt
3、检查磁盘文件d:/test.txt是否存在，如果存在，则删除

通过“Task分配线程”分配Task，再结合线程池，可以提高效率。存在空闲线程则从Queue取出一个Task分配到该线程上。其实就是通过分配实现并行。当然使用队列还有很多额外的好处，当然这是另一个话题了。

若果你的想实现并行的是有序操作集，则可自己建立TaskQueue作为操作集，同时作为并行单元进行任务。

17 楼 ymkyve 2011-03-14  

cjnetwork 写道

sunheavenvan 写道

个人感觉把一个Thread看作Task，只是把Task粒度放大了而已，不过真的，与Task有何区别？若果你需要发送有序的操作集合，干嘛不用队列呢？队列=有序集合，如发送的是TaskQueue如何？

当然不一定就必须使用线程作为任务的载体，也可以使用Task，或是别的（如设计一个接口，在客户端复原对象的时候，强制转换被传输对象为接口，然后调用接口方法）。不过个人感觉使用线程还是有一定的优势的，这种分布式本来就是处理比较耗时的任务（或是叫做工作），当这样的工作发送给客户端执行的时候，客户端再接收到这样的任务的时，一定不能以阻塞式的执行这样一个任务，然后返回结果给服务器端，因为服务器端可能源源不断在传输这样的多个任务到客户端，并且这些任务不具有任何联系，各自完成一部分逻辑操作。客户端再接收到这样的工作后，需要单独开启线程资源来执行这些任务，并行的。

使用队列这一点，你能讲的具体一点么？我这里没有怎么理解，使用队列能满足这样的需求么，如何实现？
假如有这样三个事情需要处理：
1、计算1---1000的累加和，保存结果到本地磁盘文件中d:/calculateResult.txt
2、下载http://www.baidu.cn网页内容,保存结果到磁盘文件d:/downloadResult.txt
3、检查磁盘文件d:/test.txt是否存在，如果存在，则删除

LZ的想法是正确的,至于前面提到的是否合并task结果,是否让部分task有序,那是后续和细化
可以为每个客户机分配一个queue,让分配task的线程往各个queue里塞任务
当然也要看客户机是怎么实现的,可以做些有序化的处理什么的
思路随便想想总有的,我也是随便说的

16 楼 cjnetwork 2011-03-14  

sunheavenvan 写道

个人感觉把一个Thread看作Task，只是把Task粒度放大了而已，不过真的，与Task有何区别？若果你需要发送有序的操作集合，干嘛不用队列呢？队列=有序集合，如发送的是TaskQueue如何？

当然不一定就必须使用线程作为任务的载体，也可以使用Task，或是别的（如设计一个接口，在客户端复原对象的时候，强制转换被传输对象为接口，然后调用接口方法）。不过个人感觉使用线程还是有一定的优势的，这种分布式本来就是处理比较耗时的任务（或是叫做工作），当这样的工作发送给客户端执行的时候，客户端再接收到这样的任务的时，一定不能以阻塞式的执行这样一个任务，然后返回结果给服务器端，因为服务器端可能源源不断在传输这样的多个任务到客户端，并且这些任务不具有任何联系，各自完成一部分逻辑操作。客户端再接收到这样的工作后，需要单独开启线程资源来执行这些任务，并行的。

使用队列这一点，你能讲的具体一点么？我这里没有怎么理解，使用队列能满足这样的需求么，如何实现？
假如有这样三个事情需要处理：
1、计算1---1000的累加和，保存结果到本地磁盘文件中d:/calculateResult.txt
2、下载http://www.baidu.cn网页内容,保存结果到磁盘文件d:/downloadResult.txt
3、检查磁盘文件d:/test.txt是否存在，如果存在，则删除

15 楼 sunheavenvan 2011-03-14  

个人感觉把一个Thread看作Task，只是把Task粒度放大了而已，不过真的，与Task有何区别？若果你需要发送有序的操作集合，干嘛不用队列呢？队列=有序集合，如发送的是TaskQueue如何？

14 楼 mathgl 2011-03-14  

pangyi 写道

楼主，有些创意。但不知道有没有类似的开源产品。

fork-join ...

douglas lea is working hard for that...included in JDK 7...

13 楼 pangyi 2011-03-14  

楼主，有些创意。但不知道有没有类似的开源产品。

12 楼 cjnetwork 2011-03-14  

peterwei 写道

那你这个task是小task.如果有一个task,很大，运行完要很久。能不能把这个task分在不同机器上同时跑，跑完合并task结果。

这种牵涉到逻辑切分的东东，暂时不会。。。。。。

11 楼 peterwei 2011-03-14  

cjnetwork 写道

peterwei 写道

我也觉得这个应该算是并行计算，多机器并行task。严格上来说，我认应该是多进程，虽然这里你用线程做最小单位。

你说是多进程，这个我同意，我觉得也可以这样理解，每个进程中可以运行多个线程，每个进程里面的线程不是在进程初始化的时候拥有的，是通过网络的传输从分发服务器端获得的，这样的过程可以持续，这样的工作端进程是可以一直运行，等待处理分发服务器端的线程任务。

“多机器并行task”
这个似乎不是这样，在分发服务器端获取到这样一个task之后，只会将这样一个任务交给一个客户端处理，并不会在多个机器上都持有这样一个对象（任务）的副本，并且，分发之后，在服务器端只是保留了这样一个对象的唯一标示（uuid），并不保留这样一个任务的任何引用，在服务器端的垃圾回收阶段，可能将这个任务的内存分配都给回收，这个对象在创建---分发之后，就只存在于某一台客户端机器上（某一个进程中)上。

那你这个task是小task.如果有一个task,很大，运行完要很久。能不能把这个task分在不同机器上同时跑，跑完合并task结果。

10 楼 cjnetwork 2011-03-14  

peterwei 写道

我也觉得这个应该算是并行计算，多机器并行task。严格上来说，我认应该是多进程，虽然这里你用线程做最小单位。

你说是多进程，这个我同意，我觉得也可以这样理解，每个进程中可以运行多个线程，每个进程里面的线程不是在进程初始化的时候拥有的，是通过网络的传输从分发服务器端获得的，这样的过程可以持续，这样的工作端进程是可以一直运行，等待处理分发服务器端的线程任务。

“多机器并行task”
这个似乎不是这样，在分发服务器端获取到这样一个task之后，只会将这样一个任务交给一个客户端处理，并不会在多个机器上都持有这样一个对象（任务）的副本，并且，分发之后，在服务器端只是保留了这样一个对象的唯一标示（uuid），并不保留这样一个任务的任何引用，在服务器端的垃圾回收阶段，可能将这个任务的内存分配都给回收，这个对象在创建---分发之后，就只存在于某一台客户端机器上（某一个进程中)上。

9 楼 cjnetwork 2011-03-14  

wandou 写道

task跟thread的确是两个概念。
线程从来都是性能的杀手。
假如客户端使用完成端口来处理这个task怎么办？

task跟thread的确是两个概念。
我可能在描述上有的时候没有说得很清楚，我将需要处理的事情也看做是一个任务，我这里讲的任务并不是java中直接使用new等方式创建的task，本文中提到的任务是指一系列的有序操作的集合。

线程从来都是性能的杀手。
线程的start才会导致系统资源的消耗，在服务器端，只是使用了new 方法来创建一个新的对象，并没有分配线程资源，这个任意别的对象的初始化是一样的，不会因为它是线程对象而对系统的性能造成影响。创建好的对象，会以流的方法分发到工作的客户端，客户端才会为这个线程消耗系统资源，调度这个线程运行。这就是设计这个分布式的理由了，利用多台机器的硬件资源得到运行结果。

假如客户端使用完成端口来处理这个task怎么办？
这个没有看懂意思。。。。

8 楼 ray_linn 2011-03-14  

.....和线程有什么关系.....

建议楼主参考一下C# 4.0 task，那个叫简单哦

7 楼 peterwei 2011-03-14  

我也觉得这个应该算是并行计算，多机器并行task。严格上来说，我认应该是多进程，虽然这里你用线程做最小单位。

6 楼 wandou 2011-03-14  

task跟thread的确是两个概念。
线程从来都是性能的杀手。
设计依赖线程，是个错误的依赖方向。
假如客户端使用完成端口来处理这个task怎么办？

5 楼 mxswl 2011-03-14  

cjnetwork 写道

mxswl 写道

发送的单位应该是task而不是thread吧，客户端应该有一个task list，再由某个线程池来处理这个list的执行。

别个那个Class之间的依赖关系处理的问题，我觉得有这几种方法。
1.强制保证客户端与服务器处于相同的基础ClassPath环境下，可以通过时间检查点来保证这一点。

2.引入诸如OSGI这种需要依赖配置的依赖管理的框架。

3.自己对Class文件进行import级别的依赖分析，这一点基本不太可能做到自动化，比如我不能过import而直接用class.forname("xxxx")，你基本很难分析出来你这个依赖于xxxx。

发送的东西是能够被序列化的对象，对象都能够以流的方式被发送，至于是task或是thread，这个都没有什么关系，重要的是在传输之后，能在客户端上复原对象，然后让对象处理对应的业务逻辑。这里之所以选择  基于线程 ，是因为觉得线程能够在客户端复原后运行，执行对应的业务逻辑操作。

至于class的依赖关系，我觉得还是不要强制同一个classpath等方式，因为在这里，就是为了方便开发，客户端根本就不需要知道服务器的任务是什么，只需要注册到服务器，由服务器分发任务即可，服务器在分发任务的时候，需要将客户不能识别的类定义传送给客户端，方便客户端接收。这样，在开发的时候，只需要关注任务是什么，不关注服务器与客户端的部署问题，客户端可以是世界上的任意机器，在任意位置，启动后注册，然后就能利用客户端的硬件资源帮助处理业务。

  

4 楼 cjnetwork 2011-03-14  

mxswl 写道

发送的单位应该是task而不是thread吧，客户端应该有一个task list，再由某个线程池来处理这个list的执行。

别个那个Class之间的依赖关系处理的问题，我觉得有这几种方法。
1.强制保证客户端与服务器处于相同的基础ClassPath环境下，可以通过时间检查点来保证这一点。

2.引入诸如OSGI这种需要依赖配置的依赖管理的框架。

3.自己对Class文件进行import级别的依赖分析，这一点基本不太可能做到自动化，比如我不能过import而直接用class.forname("xxxx")，你基本很难分析出来你这个依赖于xxxx。

发送的东西是能够被序列化的对象，对象都能够以流的方式被发送，至于是task或是thread，这个都没有什么关系，重要的是在传输之后，能在客户端上复原对象，然后让对象处理对应的业务逻辑。这里之所以选择  基于线程 ，是因为觉得线程能够在客户端复原后运行，执行对应的业务逻辑操作。

至于class的依赖关系，我觉得还是不要强制同一个classpath等方式，因为在这里，就是为了方便开发，客户端根本就不需要知道服务器的任务是什么，只需要注册到服务器，由服务器分发任务即可，服务器在分发任务的时候，需要将客户不能识别的类定义传送给客户端，方便客户端接收。这样，在开发的时候，只需要关注任务是什么，不关注服务器与客户端的部署问题，客户端可以是世界上的任意机器，在任意位置，启动后注册，然后就能利用客户端的硬件资源帮助处理业务。

3 楼 mxswl 2011-03-14  

发送的单位应该是task而不是thread吧，客户端应该有一个task list，再由某个线程池来处理这个list的执行。

别个那个Class之间的依赖关系处理的问题，我觉得有这几种方法。
1.强制保证客户端与服务器处于相同的基础ClassPath环境下，可以通过时间检查点来保证这一点。

2.引入诸如OSGI这种需要依赖配置的依赖管理的框架。

3.自己对Class文件进行import级别的依赖分析，这一点基本不太可能做到自动化，比如我不能过import而直接用class.forname("xxxx")，你基本很难分析出来你这个依赖于xxxx。

2 楼 cjnetwork 2011-03-14  

scamperdog 写道

和线程有什么关系？

将需要处理的工作封装在线程内，由服务器将这个工作线程分发给客户端处理，这就是和线程的关系。

如果不适用线程，没有想到好一点的方法将一个工作（任务）从一个机器传到另外一个机器上，然后怎么样复原。