主仆模式(Master-Slave)
2017-02-23 11:51
525 查看
十、主仆模式(Master-Slave)
1、核心思想
一个基于分而治之思想设计模式,将一个任务(原始任务)分解为若干个语义等同的子任务,
并由专门的工作者线程来并行执行这些任务,原始任务的结果是通过整合各个子任务的处理结果形成的。
2、评价:
即提高计算效率,又实现了信息隐藏。
3、适用场景
a、并行计算,目的是提升计算性能。
b、容错处理,目的是提高计算的可靠性,因为原始任务的结果是通过整合各个子任务的处理结果形成的
c、计算精度,目的是提高计算的精确程度,因为原始任务的结果是所有子任务的处理结果不精确性最低的一个结果。
只有a场景中Slave参与者实例是同一个实现类,b、c场景不同Slave参与者对应着不同的实现类。使用策略模式继承同一个接口
4、收集子任务处理结果的2种方法
a Master和Slave使用同一个存储仓库
b 使用Future模式
5、可靠性与异常处理
可以捕获子任务异常,可以考虑由其自身重新执行处理失败的子任务,即在客户端线程执行。
1、核心思想
一个基于分而治之思想设计模式,将一个任务(原始任务)分解为若干个语义等同的子任务,
并由专门的工作者线程来并行执行这些任务,原始任务的结果是通过整合各个子任务的处理结果形成的。
2、评价:
即提高计算效率,又实现了信息隐藏。
3、适用场景
a、并行计算,目的是提升计算性能。
b、容错处理,目的是提高计算的可靠性,因为原始任务的结果是通过整合各个子任务的处理结果形成的
c、计算精度,目的是提高计算的精确程度,因为原始任务的结果是所有子任务的处理结果不精确性最低的一个结果。
只有a场景中Slave参与者实例是同一个实现类,b、c场景不同Slave参与者对应着不同的实现类。使用策略模式继承同一个接口
4、收集子任务处理结果的2种方法
a Master和Slave使用同一个存储仓库
b 使用Future模式
5、可靠性与异常处理
可以捕获子任务异常,可以考虑由其自身重新执行处理失败的子任务,即在客户端线程执行。
/** * @author huzhiqiang * @param <T> 子任务类型 */ public interface TaskDivideStrategy<T> { /** * 返回下一个子任务,返回值为null,则表示无后续子任务 * @return */ T nextChunk(); } /** * 对子任务派发算法策略的抽象 * @author huzhiqiang * * @param <T> 子任务类型 * @param <V> 子任务处理结果类型 */ public interface SubTaskDispatchStrategy<T, V> { /** * 根据指定的原始任务分解策略,将分解得来的各个子任务派发给一组Slave参与者实例 * @param slaves * @param taskDivideStrategy 原始任务分解策略 * @return * @throws InterruptedException */ Iterator<Future<V>> dispatch(Set<? extends SlaveSpe<T, V>> slaves, TaskDivideStrategy<T> taskDivideStrategy) throws InterruptedException; } /** * Master-Slave模式Slave参与者的抽象 * @author huzhiqiang * * @param <T> 子任务类型 * @param <V> 子任务处理结果类型 */ public interface SlaveSpe<T, V> { public Future<V> submit(final T task) throws InterruptedException; public void init(); public void shutdown(); } /** * 对处理失败的子任务进行重试所需的信息 * @author huzhiqiang * * @param <T> * @param <V> */ public class RetryInfo<T, V> { public final T subTask; public final Callable<V> redoCommand; public RetryInfo(T subTask, Callable<V> redoCommand) { super(); this.subTask = subTask; this.redoCommand = redoCommand; } } public class SubTaskFailureException extends Exception { private static final long serialVersionUID = 1L; @SuppressWarnings("rawtypes") public final RetryInfo retryInfo; @SuppressWarnings("rawtypes") public SubTaskFailureException(RetryInfo retryInfo, Exception cause) { super(cause); this.retryInfo = retryInfo; } } /** * 简单的轮询派发算法策略 * @author huzhiqiang * * @param <T> * @param <V> */ public class RoundRobinSubTaskDispatchStrategy<T, V> implements SubTaskDispatchStrategy<T, V> { @Override public Iterator<Future<V>> dispatch(Set<? extends SlaveSpe<T, V>> slaves, TaskDivideStrategy<T> taskDivideStrategy) throws InterruptedException { final List<Future<V>> subResults = new LinkedList<Future<V>>(); T subTask; Object[] arrSlaves = slaves.toArray(); int i = -1; final int slaveCount = arrSlaves.length; Future<V> subTaskResultPromise; //先划分子任务,然后通过轮询算法分配到子任务处理线程中 //如果子任务处理线程不是同一个实例对象,则划分子任务时可做对应处理 while(null != (subTask = taskDivideStrategy.nextChunk())){ i = (i+1) % slaveCount; subTaskResultPromise = ((WorkerThreadSlave<T, V>) arrSlaves[i]).submit(subTask); subResults.add(subTaskResultPromise); } return subResults.iterator(); } } /** * 基于工作者线程的Slave参与者通用实现 * @author huzhiqiang * * @param <T> * @param <V> */ public abstract class WorkerThreadSlave<T, V> extends AbstractTerminatableThread implements SlaveSpe<T, V> { private final BlockingQueue<Runnable> taskQueue; public WorkerThreadSlave(BlockingQueue<Runnable> taskQueue) { super(); this.taskQueue = taskQueue; } @Override public Future<V> submit(T task) throws InterruptedException { FutureTask<V> ft = new FutureTask<V>(new Callable<V>() { @Override public V call() throws Exception { V result; try { result = doProcess(task); } catch (Exception e) { SubTaskFailureException ex = newSubTaskFailureException(task, e); throw ex; } return result; } }); taskQueue.put(ft); terminationToken.reservations.incrementAndGet(); return ft; } private SubTaskFailureException newSubTaskFailureException(final T subTask, Exception e) { RetryInfo<T, V> retryInfo = new RetryInfo<T, V>(subTask, new Callable<V>() { @Override public V call() throws Exception { V result; result = doProcess(subTask); return result; } }); return new SubTaskFailureException(retryInfo, e); } @Override public void init() { start(); } @Override public void shutdown() { //会等待线程处理完成再返回 terminate(true); } @Override protected void doRun() throws Exception { try { Runnable task = taskQueue.take(); //不能使用new Thread(task).start() task.run(); } finally{ terminationToken.reservations.decrementAndGet(); } } /** * 用于实现子任务的处理逻辑 * @param task * @return * @throws Exception */ public abstract V doProcess(T task) throws Exception; } public abstract class AbstractMaster<T, V, R> { public volatile Set<? extends SlaveSpe<T, V>> slaves; //子任务派发算法策略 public volatile SubTaskDispatchStrategy<T, V> dispatchStrategy; public AbstractMaster(){} public void init(){ slaves = createSlaves(); dispatchStrategy = newSubTaskDispatchStrategy(); for(SlaveSpe<T, V> slave: slaves){ slave.init(); } } /** * 对子类暴露的服务方法,该类的子类需要定义一个比该方法命名更为具体的服务方法 * 该方法用了 Template(模板)模式 strategy(策略)模式 * @param params * @return * @throws Exception */ public R service(Object... params) throws Exception{ final TaskDivideStrategy<T> taskDivideStrategy = newTaskDivideStrategy(params); Iterator<Future<V>> subResults = dispatchStrategy.dispatch(slaves, taskDivideStrategy); for(SlaveSpe<T, V> slave: slaves){ slave.shutdown(); } //slave.shutdown() 会等待线程处理完成再运行 R result = combineResults(subResults); return result; } private SubTaskDispatchStrategy<T, V> newSubTaskDispatchStrategy() { return new RoundRobinSubTaskDispatchStrategy<T, V>(); } //留给子类实现, 用于合并子任务处理结果 public abstract R combineResults(Iterator<Future<V>> subResults); //留给子类实现, 用于创建原始任务分解算法策略 public abstract TaskDivideStrategy<T> newTaskDivideStrategy(Object... params); //留给子类实现, 用于创建Slave参与者实例 public abstract Set<? extends SlaveSpe<T, V>> createSlaves(); } public class Range { public final int lowerBound; public final int upperBound; public Range(int lowerBound, int upperBound) { if(upperBound < lowerBound){ throw new IllegalArgumentException("upperBound should not be less than lowerBound"); } this.lowerBound = lowerBound; this.upperBound = upperBound; } @Override public String toString() { return "Range [lowerBound=" + lowerBound + ", upperBound=" + upperBound + "]"; } } /** * 质数生成器服务。模式角色: Master-Slave.Master */ public class PrimeGeneratorService extends AbstractMaster<Range, Set<BigInteger>, Set<BigInteger>> { public PrimeGeneratorService(){ this.init(); } //对外提供的具体方法名 public Set<BigInteger> generatePrime(int upperBound) throws Exception{ return this.service(upperBound); } @Override public Set<BigInteger> combineResults(Iterator<Future<Set<BigInteger>>> subResults) { Set<BigInteger> result = new TreeSet<BigInteger>(); while(subResults.hasNext()){ try { result.addAll(subResults.next().get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { Throwable cause = e.getCause(); if(SubTaskFailureException.class.isInstance(cause)){ @SuppressWarnings("rawtypes") RetryInfo retryInfo = ((SubTaskFailureException) cause).retryInfo; Object subTask = retryInfo.subTask; System.out.println("failed subTask: " + subTask); } } } return result; } @Override public TaskDivideStrategy<Range> newTaskDivideStrategy(Object... params) { final int numOfSlaves = slaves.size(); final int originalTaskScale = (int) params[0]; final int subTaskScale = originalTaskScale / numOfSlaves; final int subTaskCount = (0 == (originalTaskScale % numOfSlaves)) ? numOfSlaves: numOfSlaves+1; TaskDivideStrategy<Range> tds = new TaskDivideStrategy<Range>() { private int i=1; @Override public Range nextChunk() { int upperBound; if(i<subTaskCount){ upperBound = i * subTaskScale; }else if(i==subTaskCount){ upperBound = originalTaskScale; }else{ return null; } int lowerBound = (i-1)*subTaskScale + 1; i++; return new Range(lowerBound, upperBound); } }; return tds; } @Override public Set<? extends SlaveSpe<Range, Set<BigInteger>>> createSlaves() { Set<PrimeGenerator> slaves = new HashSet<>(); for(int i=0; i<Runtime.getRuntime().availableProcessors(); i++){ slaves.add(new PrimeGenerator(new ArrayBlockingQueue<Runnable>(200))); } return slaves a708 ; } /** * 质数生成器。模式角色: Master-Slave.Slave */ private static class PrimeGenerator extends WorkerThreadSlave<Range, Set<BigInteger>>{ public PrimeGenerator(BlockingQueue<Runnable> taskQueue){ super(taskQueue); } @Override public Set<BigInteger> doProcess(Range range) throws Exception { Set<BigInteger> result = new TreeSet<BigInteger>(); BigInteger start = BigInteger.valueOf(range.lowerBound); BigInteger end = BigInteger.valueOf(range.upperBound); //java.math.BigInteger.nextProbablePrime() 此方法返回一个整数大于该BigInteger的素数 while(-1 == (start = start.nextProbablePrime()).compareTo(end)){ result.add(start); } return result; } } } public class Main { public static void main(String[] args) throws Exception { for(int i=0;i<10;i++){ PrimeGeneratorService service = new PrimeGeneratorService(); Set<BigInteger> result = null; try { result = service.generatePrime(Integer.valueOf(1000*(i+1))); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } System.out.println("Generated " + result.size() + " prime:"); System.out.println(result.toString()); } } }
相关文章推荐
- MySQL数据库集群Master-Slave模式安装摘要
- mysql模式:master/slave
- redis 学习笔记(3)-master/slave(主/从模式)
- tns load balance及master/slave使用模式和灰度发布
- redis-master/slave模式
- MongoDB配置Master和Slave模式
- redis -master/slave(主/从模式)
- redis 学习笔记(3)-master/slave(主/从模式)
- MySQL master/slave 模式
- MongoDB的Master-Slave主从模式配置及主从复制要点解析
- mysql模式:master/slave(二)
- Mongodb的master-slave模式与master-master模式实验
- MPI master & slave 模式的基本框架
- redis -master/slave(主/从模式)
- solr replication,大部分master-slave模式才能用
- mysql的master-slave模式
- 单PC Windows环境MySQL主从master-slave模式配置与测试
- mysql master/slave 模式
- Redis_master-slave模式
- 多线程设计模式——Master-Salave(主仆)模式