多线程并行执行及调度管理-两种实现方式

利用多线程可以通过并行执行任务来提升效率，但是很多场景下，不是所有的任务都是可以一起执行的，现实情况是有的任务必须要等到之前那些可并行的任务都执行后才可以继续执行的。考虑如下任务场景：

任务一和任务二可并行执行，但是任务三必须等到一、二执行完后才能执行，任务四执行完后任务五、六、七才能再并行，一般我们的实现方式是按阶段顺序处理，在可并行处利用多线程手段，但是这种方式往往依赖任务的特点，下面我提供可串并新的框架，根据任务的配置灵活实现并行处理及串行保证，此框架我将会利用jdk7引入的fork-join及akka并行处理框架来分别实现。

 

实现思路：

将要执行的任务抽象一下，引入一个标记参数isAsyn来表示是否可异步执行，如果true，表示当前任务可以与之前环节的任务并行执行，如果为false，表示当前任务为串行任务，必须等到之前的任务直接完成后才可以执行。引入此标记后，上述任务场景参数分别为：

任务一(isAsyn=true)，任务二(isAsyn=true)，任务三(isAsyn=false)，任务四(isAsyn=false)，任务五(isAsyn=true)，任务六(isAsyn=true)，任务七(isAsyn=true)。此场景的任务按先后组合在一起即为整个任务链，并行处理框架得到此场景的任务链后，从该场景中解析分辨出可并行和需要串行的任务子链，对可并行的任务链利用多线程执行，对需串行的任务链需要在执行前进行一下判断，即如果有它之前的任务链正在处理，则等待其执行完成，否则执行此串行链。

 

实现方式一：ForkJoin实现

1. 将任务放入到待处理链中，所谓的链其实是一个List:

2. 1. List<AbstractProcessor> processorList;
      
      //初始化processorList后调用addProcessor放入任务
      publicvoid addProcessor(AbstractProcessor processor)throwsException{
          this.addProcessor(this.processorList.size(), processor);
      }
      

      　　

   2. 初始化ForkJoinPool，并创建一个继承自 RecursiveAction  的类，以便ForkJoinPool调用：

      privatestaticfinalForkJoinPool forkJoinPool =newForkJoinPool();
      
      // forkjoin框架执行类
      classProcessorActionextendsRecursiveAction{
          privatestaticfinallong serialVersionUID =-176563833918642543L;
          privatefinalProcessorChain pChain;
          //参数即待处理的任务链
          ProcessorAction(ProcessorChain pChain){
              this.pChain = pChain;
          }
          @Override
          protectedvoid compute(){
              // 见下文
          }
      }        
      

      　　

3. 调用 forkJoinPool.invoke(new ProcessorAction(processorChain));  进行处理，此方法会自动调用 ProcessorAction  类的compute方法，串并行的关键在于此方法的内部实现。
4. 以下是compute方法的实现：

   1. 		@Override
      		protected void compute() {
      			AbstractProcessor processor = null;
      			if (pChain.size() == 1) {
      				try {
      					//如果此子链中只有一个任务了，开始执行此任务
      					processor = pChain.nextProcessor(); // 取出任务
      					processor.process(processContext);
      				} catch (Exception e) {
      					this.completeExceptionally(e);
      				}
      			} else {
      				// 否则，说明此链包含多个任务，则需要判断哪些是可并行的，哪些是只能串行的
      				List<AbstractProcessor> asynProcessorList = new ArrayList<>();
      				while (null != (processor = pChain.nextProcessor())) {
      					if (processor.isASyn()) {
      						// 循环遍历链中的任务，如果是可并行的，放到并行任务列表中
      						asynProcessorList.add(processor);
      					} else {
      						// 如果这个任务是串行的，则需要分情况进行处理
      						if (!asynProcessorList.isEmpty()) {
      							// 情况一是此任务之前有待并行的任务，则需要将那些任务先执行
      							invokeAllProcessor(asynProcessorList);
      							ForkJoinTask.helpQuiesce(); // 等待之前的任务执行完成
      						}
      						// 处理完之前的任务后开始处理自己
      						ProcessorAction synProcessorAction = new ProcessorAction(pChain.buildNewProcessorChain(processor));
      						synProcessorAction.invoke(); // 等待自己处理完成
      						// 此并行任务处理完后，将之后的任务当成一个新的任务链，重复上述过程
      						ProcessorAction newProcessorAction = new ProcessorAction(pChain.subConfigProcessorChain(pChain.getIndex(processor)+1, pChain.size()));
      						newProcessorAction.fork();
      						ForkJoinTask.helpQuiesce();// 等待新的任务链执行完成
      						asynProcessorList.clear(); // 情况异步任务列表
      						break;
      					}
      				}
      				if (!asynProcessorList.isEmpty()) {
      					// 如果没有遇到可串行任务，说明当前任务链都是可并行的，则全部并行
      					invokeAllProcessor(asynProcessorList);
      				}
      			}
      		}
      
      
      		private void invokeAllProcessor(List<AbstractProcessor> asynProcessorList) {
      			List<ProcessorAction> taskList = new ArrayList<>();
      			for(AbstractProcessor processor : asynProcessorList){
      				taskList.add(new ProcessorAction(pChain.buildNewProcessorChain(processor)));
      			}
      			invokeAll(taskList);
      		}
      

      　　

实现方式二：AKKA实现

AKKA的实现原理与forkjoin类似，只不过akka利用消息机制，在各处理线程间进行通信，其效率更高，通过实际测试也可以看到，利用akka的实现比forkjoin实现速度更快。建议有类似并行控制处理需求的优先选择akka，值得注意的是akka是scala实现的，调试起来有一点小麻烦。以下给出主要代码及注释：

1. 创建执行system及master：

   1. privatestaticActorSystem execSystem =ActorSystem.create("processorHandleSystem");
      masterRef = execSystem.actorOf(Props.create(ProcessMaster.class,newCreator<ProcessMaster>(){
      privatestaticfinallong serialVersionUID =-7303628261382312794L;
      @Override
      publicProcessMaster create()throwsException{
          returnnewProcessMaster(processorChain, processContext, processInfo);
      }
      }),"master"+ UUID.randomUUID());
      

      　　
    
2. ProcessMaster是任务分发类，也是整个任务链执行的开始：

   private final static class ProcessMaster extends UntypedActor {
   		private ProcessorChain pChain;
   		private Set<ActorRef> actorRefSet;
   		private ActorRef startSender ; //保存启动方法的sender
   		ProcessMaster(ProcessorChain processChain) {
   			this.pChain = processChain;
   			this.actorRefSet = new HashSet<>();
   		}
   		@Override
   		public void onReceive(Object message) throws Exception {
   			if (message instanceof String) {
   				if ("doProcessor".equals(message)) {
   					// 如果是master自己告訴的消息，說明要开始干活了
   					tellWorkerDoProcessor(actorRefSet);
   					startSender = getSender();
   				} else if ("workDone".equals(message)) {
   					// worker执行完后发送workDone消息
   					if (null == this.pChain.getNextProcessor()) {
   						// 如果链里面没任务了，告诉master任务做完了
   						startSender.tell("finished", getSelf());
   						this.getContext().stop(getSelf());
   					} else {
   						// 如果还有任务，则说明还有其他并行的任务还在做，则在set里面先把自己清除掉
   						actorRefSet.remove(this.getSender());
   						if (actorRefSet.isEmpty()) {
   							// 此set为空，表示所以并行的任务做完了，可以到下一阶段继续做任务了
   							tellWorkerDoProcessor(actorRefSet);
   						}
   					}
   				}
   			} else if (message instanceof Exception) {
   				startSender.tell(message, getSelf());
   				getContext().stop(getSelf());
   			} else {
   				unhandled(message);
   			}
   		}
   

   　　

    
3. tellWorkerDoProcessor即是告诉任务开始执行的方法：

   private void tellWorkerDoProcessor(Set<ActorRef> actorRefSet) {
   			System.out.println(AbstractExecutor.getPChainContent(pChain));
   			AbstractProcessor abstractProcessor = pChain.getNextProcessor();
   			do {
   				if (null == abstractProcessor) {
   					// 保险起见，可以不加
   					break;
   				}
   				if (!abstractProcessor.isASyn() && !actorRefSet.isEmpty()) {
   					// 如果当前需要同步，但是set不为空，说明之前的异步任务没做完，需要返回等到那些任务做完
   					break;
   				}
   				abstractProcessor = pChain.nextProcessor();
   				// 准备好一个带处理的任务
   				ActorRef workActorRef = this.getContext().actorOf(Props.create(ProcessWorker.class, new Creator<ProcessWorker>() {
   					private static final long serialVersionUID = -8655672550330372007L;
   					@Override
   					public ProcessWorker create() throws Exception {
   						return new ProcessWorker();
   					}
   				}));
   				workActorRef.tell(abstractProcessor, this.getSelf()); // 告诉此任务开始执行了
   				actorRefSet.add(workActorRef);
   				if (!abstractProcessor.isASyn()) {
   					// 如果当前的任务是需要同步的，则调处while循环，没必要找下一个了，这个同步的必须先执行完
   					break;
   				}
   			} while (null != (abstractProcessor = pChain.getNextProcessor()));
   		}
   

   　　

4. ProcessWorker是真正任务执行的地方：

   // 处理Processor的类
   	private final static class ProcessWorker extends UntypedActor {
   		public ProcessWorker() {
   		}
   		@Override
   		public void onReceive(Object message) throws Exception {
   			if (message instanceof AbstractProcessor) {
   				AbstractProcessor processor = (AbstractProcessor) message;
   				System.out.println("开始处理processor：" + processor);
   				try {
   					processor.process();
   					this.getSender().tell("workDone", this.getSelf()); 
   				} catch (Exception e) {
   					this.getSender().tell(e, this.getSelf());
   				}
   			}
   		}
   	}
   

   　　以上即是两种方式的不同实现，从效率上看，基于消息机制的akka比基于线程间通信的forkjoin效率更好，在并行及串行任务混杂的场景里，可以考虑利用akka来实现，提升效率。

 

题外话：

自从来商城后有三个多月没有文章了，打破一个习惯比养成一个习惯容易得多了，很多时候文章不是为了给大家看，而是对自己的一个总结和反省，毕竟接触的东西多了，容易杂而不精，只有通过不同的总结提炼，将他们概括成通用的可记忆的知识，才能融会贯通，更好内容关注后续吧。