当前位置:首页 > 开发 > 系统架构 > 架构 > 正文

大规模分布式系统架构与设计实战笔记1

发表于: 2014-08-01   作者:臻是二哥   来源:转载   浏览:
摘要: 好久没有写博客了,这段时间一直在读一本书《大规模分布式系统架构与设计实战》,这是淘宝千峰老师的作品,讲了一个分布式框架fourinone,我是吧这本书作为我走向分布式的启蒙老师的。 fourinone是一个分布式框架,它主要就是一种“包工头-职介所-工人”的结构,下面我们来说书中的一个例子,使用“并行计算实现上亿排序”。 对于上百G甚至上T的数据,这种数据量是巨大的,通常这些数据也无法在一台计
好久没有写博客了,这段时间一直在读一本书《大规模分布式系统架构与设计实战》,这是淘宝千峰老师的作品,讲了一个分布式框架fourinone,我是吧这本书作为我走向分布式的启蒙老师的。
fourinone是一个分布式框架,它主要就是一种“包工头-职介所-工人”的结构,下面我们来说书中的一个例子,使用“并行计算实现上亿排序”。
对于上百G甚至上T的数据,这种数据量是巨大的,通常这些数据也无法在一台计算机上保存,这就使得数据本身就会分布在多个计算机上,即分布式存储。
现在假设我们有50G的无序数据等待排序,这50G的数据都是10万以内的数据。

当使用分布式计算时,首先我们获得10台计算机,这样每台计算机上有5G的数据,由于现在有10台计算机,因此我们考虑可以将所有的数据分成10组(正好对应10台计算机,第一台保存0~10000,第二台保存10000~20000,。。。,第十台保存90000~100000),然后每个计算机对自己的一组数据进行排序,但这样我们需要在分布式系统中有许多次的交互,因此我们想到可以想让每台计算机将自己的数据分成10组,然后再将对应的组发到对应的计算机上,这样,10台计算机之间有很好的并行性。

因此,最终我们的做法是“先在每台计算机上对数据进行分组,然后将数据合并到他们应该待得计算机上(比如每台计算机的第三组合并到第3台计算机上),最后,每台计算机上对自己的数据进行排序”,这样就实现了大规模数据的排序。当然对于分组时还可以行进行。
工人类的代码如下:
import com.fourinone.MigrantWorker;
import com.fourinone.WareHouse;
import com.fourinone.Workman;
import java.util.List;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Random;
import java.util.Collections;

public class SortWorker extends MigrantWorker
{
[color=red]//变量rammax,totalmax,total,rad用来生成需要排序的数据,[/color]
	private final int rammax;//1000000 排序数据的取值范围最大值
	private int totalmax;//排序数据的总量(每个工人)
	private int total = 0;//记录当前生成随机数据的数量
	private Random rad = new Random();


	private HashMap<Integer,List<Integer>> wharr = new HashMap<Integer,List<Integer>>();//用来数据信息,组别,以及包含的数字链表
	private int wknum=-1;
	private int index=-1;
	private Workman[] wms = null;

	public SortWorker(int totalmax, int rammax)
	{
		this.totalmax = totalmax;
		this.rammax = rammax;
	}
[color=red]//getNumber用来返回数字的所属于的组别以及数字本身[/color]
	
	public Integer[] getNumber()
	{
		if(total++<totalmax){
			int thenum = rad.nextInt(rammax);
			int numi = (thenum*wknum)/rammax;
			return new Integer[]{numi,thenum};
		}
		else return new Integer[]{-1,-1};
	}
		
	public WareHouse doTask(WareHouse wh)
	{

		int step = (Integer)wh.getObj("step");[color=red]//获取任务类别[/color]

		if(wms==null){
			wms = getWorkerAll();
			wknum = wms.length;
		}
		index = getSelfIndex();
		System.out.println("wknum:"+wknum+";step:"+step);
		WareHouse resultWh = new WareHouse("ok",1);
		
		if(step==1){
			Integer[] num = null;
			while(true){
				num = getNumber();
				if(num[0]!=-1){
					List<Integer> arr = wharr.get(num[0]);//取出该分类的list
					if(arr==null)
						arr = new ArrayList<Integer>();
					arr.add(num[1]);
					wharr.put(num[0], arr);
				}
				else break;
			}
		}else if(step==2){
			for(int i=0;i<wms.length;i++){
				if(i!=index&&wharr.containsKey(i)){
					List<Integer> othernum = wharr.remove(i);
										System.out.println(i+"-receive:"+wms[i].receive(new WareHouse(i, othernum)));//将不属于自己的分类发给其他工人
				}
			}
		}else if(step==3){
			List<Integer> curlist = wharr.get(index);
			Collections.sort(curlist);//对属于自己分类的数据进行内存内排序
			System.out.println(curlist);
			System.out.println(curlist.size());
			resultWh.setObj("total",curlist.size());
		}	
		return resultWh;
	}
	[color=red]//覆盖receive函数以实现系统节点之间的交互的功能[/color]
	protected boolean receive(WareHouse inhouse)
	{
		List<Integer> thisnum = wharr.get(index);
		thisnum.addAll((List<Integer>)inhouse.get(index));
		return true;
	}
	
	public static void main(String[] args)
	{
		SortWorker mw = new SortWorker(Integer.parseInt(args[2]),Integer.parseInt(args[3]));
		mw.waitWorking(args[0],Integer.parseInt(args[1]),"SortWorker");
	}
}

import com.fourinone.Contractor;
import com.fourinone.WareHouse;
import com.fourinone.WorkerLocal;
import java.util.Date;

public class SortCtor extends Contractor
{
	public WareHouse giveTask(WareHouse wh)
	{
		WorkerLocal[] wks = getWaitingWorkers("SortWorker");
		System.out.println("wks.length:"+wks.length+";"+wh);
		
		wh.setObj("step", 1);//1:group;
		doTaskBatch(wks, wh);
		
		wh.setObj("step", 2);//2:merge;
		doTaskBatch(wks, wh);
		
		wh.setObj("step", 3);//3:sort
		WareHouse[] hmarr = doTaskBatch(wks, wh);
		int total=0;
		for(int i=0;i<hmarr.length;i++){
			Object num = hmarr[i].getObj("total");
			if(num!=null)
				total+=(Integer)num;
		}
		wh.setObj("total",total);
		return wh;
	}
	
	public static void main(String[] args)
	{
		Contractor a = new SortCtor();
		WareHouse wh = new WareHouse();
		long begin = (new Date()).getTime();
		a.doProject(wh);
		long end = (new Date()).getTime();
		System.out.println("total:"+wh.getObj("total")+",time:"+(end-begin)/1000+"s");
	}
}

大规模分布式系统架构与设计实战笔记1

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号