330      阿里云  技术社区[云栖]

多线程处理写法

1.rxjava，observeOn(Schedulers.io())封装多线程，

Flowable.fromArray(ips)

.observeOn(Schedulers.io())

.flatMap(

ip -> Flowable.fromCallable(

() -> {

Boolean check = check(ip);
                            checkResult.put(ip,check);
                            return check;
                        }

)

).blockingSubscribe();
return checkResult;

2.线程池, 在线程池中执行saltStackUtil.restartJavaSync方法的同步，其返回值放入执行consummer

private static final ExecutorService executorService = Executors.newFixedThreadPool(10, new ThreadFactory() {
@Override
    public Thread newThread(Runnable r) {

Thread thread = new Thread(r);
        thread.setName("javaOperator");
        return thread;
    }

});

public void restartJava(String ip, String projectName, String commitId, String jarName, String jvmArgs, Consumer<AppServerOperateResult> appServerOperateResultConsumer) throws SaltException {
executorService.execute(() ->{

AppServerOperateResult appServerOperateResult = new AppServerOperateResult();
        appServerOperateResult.setIp(ip);
        appServerOperateResult.setProjectName(projectName);
        appServerOperateResult.setCommitId(commitId);
        appServerOperateResult.setJarName(jarName);
        appServerOperateResult.setStartTime(new Date());
        appServerOperateResult.setStatus(Constant.JAVA_APP_OPERATE_PROCESS);
        appServerOperateResult.setType(Constant.JAVA_APP_OPERATE_RESTART);
        appServerOperateResult.setRunParamater(jvmArgs);
        Map<String, Result<Map<String, State.ApplyResult>>> stringResultMap = null;
        try {

stringResultMap = saltStackUtil.restartJavaSync(ip, projectName, commitId, jarName, jvmArgs);
            processResult(stringResultMap,appServerOperateResult);
            appServerOperateResultConsumer.accept(appServerOperateResult);
        } catch (Exception e) {

MyExceptionHandler.handlerException(e,logger);
        }


});

}

3.stream，parallel()并行流执行foreach，consummer封装返回数据到checkResult Map中

Stream.of(ips).filter(ip -> ip!=null).parallel().forEach(new Consumer<String>() {
@Override
    public void accept(String ip) {
checkResult.put(ip, check(ip, appId, appTypeCode));
    }

});

return checkResult;

最后更新：2017-09-27 15:04:37

  上一篇：  2017 VQA Challenge 第一名技术报告

  下一篇：  目标检测101：一文带你读懂深度学习框架下的目标检测