Thread_跨节点集合查询

项目中数据库进行了水平切分，为了处理跨节点集合查询，采用了多线程并发操作的方式来处理，并且对各线程执行的结果进行操作，如果是返回结果集，则合并排序；如果是聚合操作，则求和。个人觉得该实现方式很有代表性，所以抽取了原型实现备忘。

 

集合处理：

 

import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;

/**
 * 集合查询
 * 
 * 场景：	数据库水平切分后,集合查询要对所有数据源进行操作;
 * 例如：	select count(*) from tablename,这个sql要广播到所有节点上执行;
 * 关键点：	对从各数据源查询得到的记录数进行求和.
 * 
 * @author charles
 *
 */
public class Service {
	private static ExecutorService exec;
	private CompletionService<Object> completionService;
	
	
	public String invoke(String name){
		//init fixedpool
		FixedConnectionPool fixedPool = new FixedConnectionPool();
		exec = fixedPool.getExec();
		completionService = fixedPool.getCompletionService();
		
		for(int i = 0; i < 10; i++){
			Task task = new Task(name);
			if (!exec.isShutdown()) {
				completionService.submit(task);
			}
		}
		
		return getResult();
	}
	
	
	private String getResult(){
		Object result = null;
		Object obj = null;
		
		for(int i = 0; i < 10; i++){
			try {
				obj = completionService.take().get();
				result = merge(obj,result);
				
			} catch (InterruptedException e) {
				System.out.println(">>>> Service InterruptedException : "+ e.getMessage());
				e.printStackTrace();
				obj = null;
				
			} catch (ExecutionException e) {
				System.out.println(">>>> Service ExecutionException : "+ e.getMessage());
				e.printStackTrace();
				obj = null;
				
			} catch(Exception e){
				System.out.println(">>>> Service Exception : "+ e.getMessage());
				e.printStackTrace();
				obj = null;
			}
		}
		
		
		if(result != null)
			return (String)result;
		else
			return "";
	}
	
	
	public String merge(Object obj, Object result) throws Exception {
		if(obj == null)
			obj = "";
		
		if(result == null)
			result = "";
		
		//System.out.println(">>>> merge : "+ obj +"\n"+ result);
		return obj +"\n"+ result;
	}
	
	
	public static void main(String[] args){
		System.out.println(">>>> Interface : "+ new Service().invoke("冯天魁"));
		
		if (!exec.isShutdown())
			exec.shutdown();
	}
}

 

 

任务:

/**
 * 任务
 * @author charles
 *
 */
public class Task implements Callable<Object> {
	private String name;
	private static int i = 0;
	
	/**
	 * i是多线程共享变量
	 */
	public synchronized Object call() throws Exception {
		i++;
		return "Hello "+ name + i +", you are in a fixedpool!";
	}
	
	public Task(String name){
		this.name = name;
	}
}

 

 

import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class FixedConnectionPool {
	private int cpuCoreNumber = 2;
	private ExecutorService exec;
	private CompletionService<Object> completionService;
	
	
	public FixedConnectionPool(){
		this.cpuCoreNumber = Runtime.getRuntime().availableProcessors();
		//this.exec = Executors.newCachedThreadPool();
		this.exec = Executors.newFixedThreadPool(cpuCoreNumber);
		this.completionService = new ExecutorCompletionService<Object>(exec);
	}
	
	
	public int getCpuCoreNumber() {
		return cpuCoreNumber;
	}


	public void setCpuCoreNumber(int cpuCoreNumber) {
		this.cpuCoreNumber = cpuCoreNumber;
	}


	public ExecutorService getExec() {
		return exec;
	}


	public void setExec(ExecutorService exec) {
		this.exec = exec;
	}


	public CompletionService<Object> getCompletionService() {
		return completionService;
	}

	
	public void setCompletionService(CompletionService<Object> completionService) {
		this.completionService = completionService;
	}
	
}

 

评论

5 楼 charles751 2012-09-20  

dacoolbaby 写道

觉得用Java进行排序效率很低。
一般是不是应该在数据库前面再建一个像memcache这样的东西？

怎么得到的java排序效率低？

你打算用memcached做数据缓存还是什么？

4 楼 dacoolbaby 2012-09-20  

觉得用Java进行排序效率很低。
一般是不是应该在数据库前面再建一个像memcache这样的东西？

3 楼 charles751 2012-09-19  

一般都是分页显示，每页显示的记录数不会很大，所以一般情况下结果集不会很大。

2 楼 charles751 2012-09-19  

zean 写道

这个问题很久以前就想过，如果多个节点返回的结果集都很大，怎么排序？

结果集大也没办法，汇总后排序才能保证排序的正确性。

1 楼 zean 2012-09-19  

这个问题很久以前就想过，如果多个节点返回的结果集都很大，怎么排序？