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

 

 

 

和线程有什么关系？

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

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

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

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

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

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

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

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

  

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

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

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

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

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

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

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