java master-尊龙游戏旗舰厅官网

master-worker模式简介

master-worker模式是非常经典的常用的一个并行计算模式，它的核心思想是2类进程协作工作：master进程和worker进程。master负责接收客户端请求，分配任务；worker负责具体处理任务。当各个worker处理完任务后，统一将结果返回给master，由master进行整理和总结。其好处是能够将一个大job分解成若干小job，并行执行，从而提高系统的吞吐量。比如流行的web server,如nginx,apache http都存在这种master-worker工作模式；离线分布式计算框架hadoop的jobtracker和tasktracker，实时流计算框架strom的nimbus和supervisor都涉及到这种思想。那么下面我们来具体分析下java master-worker模式的实现。

master-worker模式分析

我们重点分析下master，worker这2个角色。

master

master需要接受client端提交过来的任务task，而且还得将task分配给worker进行处理，因此master需要一个存储来存放task。那么采用哪种存储集合呢？首先来说，需要支持并发的集合类，因为多个worker间可能存在任务竞争，因此我们需要考虑java.util.concurrent包下的集合。这里可以考虑采用非阻塞的concurrentlinkedqueue。

master需要清楚的知道各个woker的基本信息，如是否各个worker都运行完毕，因此master端需要保存worker的信息，可以采用map存储。

由于最后各个worker都会上报运行结果，master端需要有一个存储结果的map，可以采用支持并发的concurrenthashmap。

worker

worker需要持有master端的任务task集合的引用，因为worker需要从里面拿取task。

同上，worker需要持有master端的存储结果的引用。

综上，我们可以得到如下：

我们可以进一步细化，master/worker应该提供什么操作？

master:

通过构造方法以初始化workers

应该提供submit(task)方法接受client端提交过来的任务

start()让workers开始处理任务

提供iscomplete()判断各个worker的状态，是否都处理完毕

提供getresult()给客户端返回结果

worker:

worker本质上就是runnable，提供run()

负责处理业务逻辑的handle()

java master-worker代码实现

task

public class task {private long id;private string name;public task(long id, string name) {this.id = id;this.name = name;}public long getid() {return id;}public void setid(long id) {this.id = id;}public string getname() {return name;}public void setname(string name) {this.name = name;}}

worker

public class worker implements runnable {private long id;private string name;private concurrentlinkedqueue workqueue;private concurrenthashmap results;public void setworkqueue(concurrentlinkedqueue workqueue) {this.workqueue = workqueue;}public void setresults(concurrenthashmap results) {this.results = results;}public worker(long id, string name) {this.id = id;this.name = name;}@overridepublic void run() {while(true){task task = workqueue.poll();if(task == null){break;}long start = system.currenttimemillis();long result = handle(task);this.results.put(task.getid(),result);system.out.println(this.name   " handle "   task.getname()   " success . result is "   result   " cost time : "   (system.currenttimemillis() - start));}}/*** 负责处理具体业务逻辑* @param task* @return*/private long handle(task task) {//这里只是模拟下，在真实环境也许是查询数据库，也许是查缓存等try {thread.sleep(500);} catch (interruptedexception e) {e.printstacktrace();}return new random().nextlong();} }

master

public class master {private concurrentlinkedqueue workqueue = new concurrentlinkedqueue();private map workers = new hashmap();private concurrenthashmap results = new concurrenthashmap();public master(int num){for(int i = 0 ; i < num ; i ){worker worker = new worker(i,"worker-"   i);worker.setresults(results);worker.setworkqueue(workqueue);workers.put(long.valueof(i),new thread(worker));}}public void submit(task task){workqueue.add(task);}public void start(){for (map.entry entry : workers.entryset()){entry.getvalue().start();}}public boolean iscomlepte(){for(map.entry entry : workers.entryset()){if(entry.getvalue().getstate() != thread.state.terminated){return false;}}return true;}public long getsumresult(){long value = 0;for(map.entry entry : results.entryset()){value = value   (long)entry.getvalue();}return value;} }

main

public class main {public static void main(string[] args) {master master = new master(10);for(int i = 0 ; i < 10 ; i ){task task = new task(i,"task-"   i);master.submit(task);}long start = system.currenttimemillis();master.start();while(true){if(master.iscomlepte()){system.out.println("sum result is "   master.getsumresult()   " . cost time : "   (system.currenttimemillis() - start));break;}}}}

运行结果

总结

在单线程的时候，处理一个task需要500ms，那么处理10个task需要5s，如果采用master-worker这种并行模型，可以大大缩短计算处理时间。

转载于:https://blog.51cto.com/zhangfengzhe/1879323

总结

以上是尊龙游戏旗舰厅官网为你收集整理的java master-worker模式实现的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得尊龙游戏旗舰厅官网网站内容还不错，欢迎将尊龙游戏旗舰厅官网推荐给好友。