Fourinone如何实现并行计算和数据库引擎

本文整理自CSDN架构师专家群技术分享内容：

关于并行计算的概念有非常多，硬件落地其实就只有两种，CPU上的并行计算和GPU上的并行计算，GPU做点积这样的矢量计算（矩阵计算）有优势，但目前还运行不了操作系统和数据库，比较多用于研究性质的计算。在我们生产系统中运用最多的是CPU上的并行计算，其落地方式也只有两种，多线程和多进程。围绕多线程、多进程结合通信技术的灵活设计，它的应用范围非常广泛，不光用于并行计算，也是大部分的服务器技术、分布式技术、中间件技术的重要实现技术。

Fourinone同时提供多线程和多进程的并行计算，并且能实现两者之间的无缝切换，而不需要改一行业务计算的逻辑代码，计算过程中还能保证高容错，一个工人故障会重投任务到其他工人执行，并有超时终止等支持。

一、Fourinone如何实现多线程并行计算
实现多线程并行计算只需要一个工头实现类就够了，除外不需要依赖任何计算服务，这个例子位于下载包的分布式计算补充demo目录下，CtorMul .java：

 

import com.fourinone.Contractor;
import com.fourinone.WareHouse;
import com.fourinone.WorkerLocal;

public class CtorMul extends Contractor
{
	public WareHouse giveTask(WareHouse inhouse)
	{
		/*WorkerLocal[] wks = getWaitingWorkers("WorkerMul");
		System.out.println("wks.length:"+wks.length);*/
		
		//生成5个多线程工人，并设置业务实现类
		WorkerLocal[] wks = getLocalWorkers(5);
		for(int j=0;j<wks.length;j++)
			wks[j].setWorker(new WorkerMul());
		
		
		//生成15个计算任务
		WareHouse[] tasks = new WareHouse[15];
		for(int i=0;i<15;i++){
			tasks[i]=new WareHouse("taskId",i+"");
		}
		
		//让多个工人并行争抢处理多个任务，并且高容错，堵塞直到返回所有任务结果。
		WareHouse[] result = doTaskCompete(wks, tasks);
		for(int i=0;i<result.length;i++){
			System.out.println(i+":"+result[i]);
		}
		
		return inhouse;
	}
	
	public static void main(String[] args)
	{
		CtorMul cd = new CtorMul();
		cd.giveTask(null);
		cd.exit();
	}
}

直接用java编译运行CtorMul.java即可完成一个本地多线程并行计算：
javac -cp fourinone.jar; *.java
java -cp fourinone.jar; CtorMul

 

我们看到CtorMul.java程序main函数里先new了一个CtorMul实例，然后调用它的giveTask方法，在这个方法里，工头自己生成一定数量的线程帮助自己完成计算任务，并在计算结束后程序退出。完成计算任务的函数是doTaskCompete，它有两个参数，把所有工人和所有任务扔给它，然后堵塞等待所有计算完成，返回一个结果集。

 

多线程并行计算方式的优缺点：
由于只需要一个工头实现类就可以完成多线程并行计算，所以它非常好集成，容易嵌入到你的业务系统中去，但是多线程并行计算有很大局限性，它首先只能在本地，无法做分布式并行计算去利用多台机器的cpu资源，还有多线程的容错性较差，由于多线程工人和主线程工头都在一个jvm进程里，一个线程出故障容易导致整个jvm进程挂掉，也比较难切换到其他线程运行。

 

二、Fourinone如何实现多进程并行计算
如何将上面的多线程并行计算无缝切换成一个多机的分布式多进程并行计算，我们把CtorMul.java里面引用到的WorkerMul.java工人实现类打开：

 

import com.fourinone.MigrantWorker;
import com.fourinone.WareHouse;

public class WorkerMul extends MigrantWorker
{
	public WareHouse doTask(WareHouse inhouse)
	{
		String taskId =  inhouse.getString("taskId");
		System.out.print("taskId"+taskId+"任务处理中...");
		try{Thread.sleep(3000L);}catch(Exception ex){}
		System.out.println("taskId"+taskId+"处理完成。");
		
		return new WareHouse("result", "ok");
	}
	
	public static void main(String[] args)
	{
		WorkerMul wd = new WorkerMul();
		wd.waitWorking(args[0],Integer.parseInt(args[1]),"WorkerMul");
	}
}

 这个工人实现类的doTask接口里，只是简单打印了任务编号和sleep了3秒模拟处理任务，然后返回。但是我们注意到它有个main函数的，上面的多线程并行计算只是new了WorkerMul 的实例作为业务实现类传入，但是并没有运行工人的main函数让它启动起来，我们可以让WorkerMul 独立启动，它就是一个工人服务进程，可以在多台机器上启动多个这样的工人进程，并监听在不同的ip和端口。

 

我们再观察到前面的CtorMul .java第一段的代码注释掉了，根据Fourinone的架构设计，我们知道工人服务进程启动后，会到职介者（ParkServer）去注册，ParkServer实现了ZooKeeper的所有功能和领导者选举算法，然后工头通过getWaitingWorkers获取到线上工人，并远程调用工人完成计算任务，详细可以参考：多机并行计算指南。

 

我们把CtorMul .java第一段的代码的注释取消，改把第二段用于本地多线程的代码注释掉，其他代码不用变。然后我们启动两个工人做多进程并行计算，重新编译后运行顺序如下：
java -cp fourinone.jar; ParkServerDemo
java -cp fourinone.jar; WorkerMul localhost 6001
java -cp fourinone.jar; WorkerMul localhost 6002
java -cp fourinone.jar; CtorMul

 

可以看到我们把并行计算从多线程切换到了多进程，但是工人的任务实现逻辑doTask仍然一行代码不改动。多进程方式需要独立运行多个工人和职介者服务，比多线程方式要麻烦和复杂，但是它能带来更强大的分布式计算扩充能力和更好的容错稳定性，我们在运行过程中，可以Ctrl+C把其中一个正在计算的工人关掉，会发现工头抛出调用异常，但是计算并未中止，而是将该任务重投到另一个工人上去做，只要集群还剩一个工人，计算就不会受影响，只是计算效率会降低，时间会延长。

 

总结：有人问Fourinone为什么不设计成Hadoop,Spark,Storm这样的动态任务投放方式，实际上Fourinone现在的方式要更灵活，如果要事先定义好DAG那样的任务流程图出来，并考虑如何分配资源，做到最后会发现都走到资源隔离管理的路上去了，那还不如开始去做Docker。从这点上看Fourinone的并行计算，更接近MPI，但是相对于MPI抽象归纳出了并行计算的角色、方法和模式。并且Fourinone也没有Hadoop和Spark的shuffle机制的烦恼，如果Fourinone做成一个资源隔离框架+DAG任务平台，那是不可能实现出一个功能强大的并行数据库引擎的，连做些灵活的机器学习算法并行化都困难，

三、再谈并行数据库引擎CoolHash
关于CoolHash，最好大家能直接它启动起来，然后各种测，与其谈太多架构算法，不如边测试边观察数据，通过数据去思考。下面是coolhash运行界面，不需要安装，启动很简单。


  

运行机器可以是普通的笔记本或者台式机，操作系统不限，CoolHash能竭尽利用你机器的最大性能，当然生产线上还是要运行在pc server上，从8核的虚拟机到24核的实体机都是可以的，不需要ssd，有更好。相信大家对相同机器配置下，关系型数据库、内存数据库的基本性能都有所了解，可以对比下性能的差异。

 

 

每个数据库厂商出示的测试报告都是宣称自己最好，所以不要去相信宣传的，比如作者亲身经历的couchdb单机能力就没有redis快，客户方把couchdb公司的人请来最后也是一样的结果。用户亲手测试的结果更有说服力，看到底能不能在单机上做到百万的TPS，达到硬件的极限，用户一定要做到自己心里有数，如果你对测试结果有什么疑问，可以直接到Fourinone技术群里去提问。

 

想强调的一点，CoolHash是持久化的，几乎同时把数据从内存刷到硬盘，所以它的容量是硬盘容量，不是内存容量，这点和redis很不一样，也就是数据容量超过内存大小时，还是运行稳定的，而且性能不下降（准确的说是不高于所在硬盘分区容量的70%-80%，操作系统建议不要超过这个安全比例），redis是严禁数据容量接近内存一半的，因为redis碰到刷硬盘的瞬间占用内存会膨胀一倍。

 

有人问，我一次写入200万数据，为什么CoolHash里只有100万，是不是丢了数据？那是因为CoolHashMap默认最大容量为1百万条数据，可根据内存大小调整HASHCAPACITY配置项，在配置文件里改下就可以了。数据库引擎是严禁有任何数据丢失错乱问题的。

 

网上有人诋毁CoolHash不支持高并发，最好的办法就是亲自检验，不用理会喷子。下载包里自带了并发测试程序：
CoolHashTestRun.java是多个客户端进程高并发大数据量读写测试
ThreadClient.java是多个客户端线程高并发大数据量读写测试
在CoolHash的最初版本只支持多进程的客户端，出于安全隔离性考虑限制了一个jvm里模拟多线程客户端访问，但是这跟数据库服务端引擎没有关系的， 引擎一直都是支持高并发访问的。现在多线程客户端也都开通了。

 

CoolHash更重要的能力还不仅是高性能读写，灵活检索能力是衡量数据库引擎能力的标志，它能支持key和value的同时模糊检索，并且可支持高并发检索，而且都是毫秒级完成，数据量比较大时才几秒。为了提升检索能力，可以设计好你的key，用点做分隔符设计一个树型key，比如user.10010.name，然后用user.*.name来检索。

 

传统关系化结构数据如何转换设计成CoolHash的树型key/value结构，可以参考下面这个图：

 除外还有key指针的设计去解决join关联的问题，CoolHash的很多特性都是其他k/v nosql数据库没有的，都来源于作者长期工作实践中的经验总结，一个追求创新的数据库引擎总是会面临有人对它的各种质疑，技术上有争论才好，被各大公司都测过，吵过，质疑过，尝试过，反思过才是对一个开源软件最好的鞭策和发展。

 

回到生产场景上，如果仅仅是从技术上验证一个东西，其实是简单的，可以先测试，再上poc项目，再上小型生产系统，可以先只做数据同步，再分担读写检索压力，运行稳定再上中型生产系统，这样一步步从小到大应用下去就很容易通过时间去检验一项技术。实际上，CoolHash在线上跑了几个月也从未出过问题，每天10-20G的数据写入（覆盖），增量2-3G，整体容量超过几百g，超出内存大小运行，一直很稳定。

 CoolHash的并行处理到底是多线程模式还是多进程模式？是多进程模式，上面是CoolHash的并行架构示意图，每个数据工人都是一个独立进程，多个jvm进程共同作业，这是因为对底层存储结构、内存、mmap等控制上，多线程操作容易出问题的，多进程更安全可靠。CoolHash是在单机上采用多进程并行模式，只有单机引擎能力上去了，整体分布式集群的能力才会提升。如果你需要基于CoolHash做分布式数据库，可以参考自带例子AsynClient.java提供的异步处理等功能支持。