Java基于线程的分布式(转自cjnetwork)

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));
        }
    }

}

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//客户端工作的线程池大小

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、内存等资源的参考，以后可以考虑在分发机制上根据工作客户端的状态进行分发

转自：http://cjnetwork.iteye.com/blog/955666