mal*_*are 5 java multithreading reactive-programming java-8 completable-future
我需要执行一些任务。有些任务是独立的，有些则依赖于其他任务的成功执行。独立的任务可以并行运行以获得更好的性能。我称这些任务为服务。该列link说明哪些服务将串行执行，哪些服务将并行执行。专栏order描述了一组定义的服务将遵循的执行顺序。对于下面的示例，服务 A 和 B 应该并行运行。如果它们已成功执行，则服务 C 将执行。请注意，服务 C 不直接依赖于其先前服务的输出，但它必须在成功执行其先前服务之后运行，因为服务 C 在其执行期间需要一些由其先前服务产生的数据。服务 C 成功执行后，下一个服务 D 将执行，如此循环下去，直到列表中的所有服务都被消费完。
Tasks service link order
Service A 01 03 1
Service B 02 03 2
Service C 03 04 3
Service D 04 05 4
Service E 05 07 5
Service F 06 07 6
Service G 07 (null) 7
以下是我的代码。
public void executeTransactionFlow(DataVo dataVo) throws Exception {
List<Callable<Boolean>> threadList = new ArrayList<>();
List<String> serviceIds = new ArrayList<>();
List<Future<Boolean>> futureList;
String validatedRespCode = null, joinTo, prevJoinTo = null, serviceId;
// Iterating through service flows map
for (Map<String, String> map : serviceFlowsMap) {
joinTo = map.get("link");
serviceId = map.get("service");
// A simple flag to differentiate which services should execute parallel and which in serial.
if (null == prevJoinTo) {
prevJoinTo = joinTo;
}
// Check for join condition. If join condition is same as previous then do not execute the thread list yet add current service in list
if (null != joinTo && joinTo.equals(prevJoinTo)) {
threadList.add(new Callable<String, DataVo>(serviceId, dataVo));
}
/*
* 1. Run the threads in the list
* 2. Empty the thread list
* 3. Empty serviceIds list
* 4. Set prevJoinTo
*/
else {
if (threadList.size() > 0) {
prevJoinTo = joinTo;
try {
// If list contain only 1 service then call, otherwise invokeAll
futureList = MyExecutor.executeServices(threadList, dataVo);
// During execution we cannot interrupt services, so we check here after they get back to here and interrupt if it has been timedout.
if (dataVo.isTimedout()) {
throw new Exception("Transaction thread is Interrupted or Timed-out");
}
// Validate service response codes and get decision in case of any failure
validatedRespCode = validateResponseOfExecutedServices(dataVo, futureList, serviceIds);
// If validationRespCode is non 00 then do not process further
if (null != validatedRespCode && !"200".equals(validatedRespCode)) {
break;
}
}
catch (Exception e) {
throw new Exception(e.getMessage(), e);
}
finally {
// clear thread list and serviceIds list. It will be populated for next parallel set of threads
threadList.clear();
serviceIds.clear();
}
}
// Start preparing new thread list
// Adding current service_id into threadList after executing previous services in parallel.
threadList.add(new Callable<String, DataVo>(serviceId, dataVo));
}
}
// Run remaining services
if (!threadList.isEmpty()) {
try {
futureList = MyExecutor.executeServices(threadList, dataVo);
validatedRespCode = validateResponseOfExecutedServices(dataVo, futureList, serviceIds);
}
catch (Throwable e) {
throw new Exception(e.getMessage(), e);
}
}
// Check validation response code
if (null != validatedRespCode && !"200".equals(validatedRespCode)) {
MyExecutor.callDeclineFlow(dataVo, validatedRespCode, null);
}
}
/**
* This method iterates through the thread list and checks for exceptions and service responses.
* If service response is not success or if any exception has occurred then exception is thrown
*/
public String validateResponseOfExecutedServices(DataVo dataVo, List<Future<Boolean>> futureList, List<String> serviceIds) throws Exception {
String finalResponse = "200", serviceResponse = null;
/*
* future list will be null if single service is executed (no other parallel transactions). The reason is that we do
* not use invokeAll() on single service.
*/
if (null != futureList && futureList.size() > 0) {
for (Future<Boolean> future : futureList) {
try {
future.get();
}
catch (Exception e) {
throw new Exception(e.getMessage(), e);
}
}
}
// Iterate through serviceIds and check responses.
for (String serviceId : serviceIds) {
serviceResponse = dataVo.getServiceResponse(serviceId);
/*
* if one of following response is found then consider it exception
*/
if (null != serviceResponse && "400,401,402,403,404,500,501".contains(serviceResponse)) {
throw new Exception("One of the service has been declined");
}
}
return finalResponse;
}
如果CompletableFuture在这里有用，那么我如何有效地使用它？
并且future.get()是阻塞调用。如果我有 10 个并行执行的服务，那么这future.get()将阻塞其他服务，即使它们在我们正在等待的当前之前执行。如何避免这种阻塞？
我添加了问题陈述的更多细节，即添加订单栏。服务需要遵循定义的顺序。服务 A 和 B 的顺序分别是 1 和 2，但它们仍然会并行执行，因为两者都有03值 in link。我认为现在不需要@Thomas 在评论中建议的基于依赖图的方法。
很棒的问题。ExecutorService虽然从技术上讲，纯粹使用and确实可以做到这一点Future，但对我来说，更好的方法是使用反应式编程，而不是纯粹依赖于FutureorCompletableFuture或CompletionService等。主要原因是它可能很快就会变成难以阅读的代码。
这是我使用RxJava 2.2.16and完成的方法ExecutorService：
ExecutorServicetosubmit()操作完成。BehaviorSubjectRxJava。操作完成后，为其每个依赖项触发步骤 (1)。ExecutorService所有操作完成后关闭。为此，请使用另一个BehaviorSubject.抱歉，由于采用了新方法，我已经用自己的方式编写了整个逻辑。但还是围绕你给出的主要要求。最好先看看中的Action模型类和createActions()方法AppRxjava。从那里，您应该能够遵循代码。为了模拟一些时间消耗，我使用了著名的Thread.sleep()技术。
public class AppRxJava{
/* To listen to the completion of a task, so that the dependent tasks may be scheduled. */
private Subject<Action> completionSub = io.reactivex.subjects.BehaviorSubject.create();
/* To listen to the completion of all tasks, so that ExecutorService may shut down. */
private Subject<Boolean> allActionCompletedSub = io.reactivex.subjects.BehaviorSubject.create();
private ExecutorService SVC = Executors.newCachedThreadPool();
private List<Action> allActions;
public static void main( String[] args ){
new AppRxJava().start();
}
private void start() {
this.allActions = createActions();
subscribeToActionCompletions();
subscribeToSvcShutdown();
startAllActions( this.allActions );
}
private void subscribeToSvcShutdown(){
/* If all actions have been completed, shut down the ExecutorService. */
this.allActionCompletedSub.subscribe( allScheduled -> {
if( allScheduled ) {
SVC.shutdown();
try {
SVC.awaitTermination( 2, TimeUnit.SECONDS );
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
private void subscribeToActionCompletions(){
this.completionSub.subscribe( complAction -> {
/* Get the actions that are dependent on this recently completed action and "attempt" to start them. */
List<Action> deps = getDeps( complAction, this.allActions );
startAllActions( deps );
/* If all actions have got completed, raise the flag. */
if( allActionsCompleted() ) this.allActionCompletedSub.onNext( true );
});
}
/* Attempts to start all actions that are present in the passed list. */
private void startAllActions( List<Action> actions ){
for( Action action : actions ) {
startAction( action, actions );
}
}
/* Attempts to start an action. Only if it is still pending and all of its dependencies are completed. */
private void startAction( Action a, List<Action> list ){
if( !a.isPending() ) return;
if( !allDepsCompleted( a, allActions ) ) return;
if( a.isPending() ) {
synchronized (a.LOCK ) {
if( a.isPending() ) {
a.setStatus( 1 ); //Set to running, so that it is not picked up twice.
SVC.submit( () -> {
try {
a.getAction().call();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
a.setStatus( 2 ); //Set to completed. (We may have to synchronize this.)
this.completionSub.onNext( a );
} );
}
}
}
}
private boolean allActionsCompleted(){
for( Action a : this.allActions ) if( !a.isCompleted() ) return false;
return true;
}
private static boolean allDepsCompleted( Action a, List<Action> allActions ){
for( Action dep : allActions ) {
if( a.getDependencies().contains( dep ) && !dep.isCompleted() ) return false;
}
return true;
}
/* Returns the actions that are dependent on Action <code>a</code>. */
private List<Action> getDeps( Action a, List<Action> list ){
List<Action> deps = new ArrayList<>();
for( Action dep : list ) if( dep.getDependencies().contains( a ) ) deps.add( dep );
return deps;
}
/* Creates the action list with respective dependencies. */
private List<Action> createActions(){
List<Action> actions = new ArrayList<>();
Action a = createAction( 5000, "ServiceA", null );
Action b = createAction( 5000, "ServiceB", null );
Action c = createAction( 2000, "ServiceC", a, b );
Action d = createAction( 2000, "ServiceD", c );
Action e = createAction( 2000, "ServiceE", d );
actions.add( a ); actions.add( b ); actions.add( c ); actions.add( d ); actions.add( e );
return actions;
}
private Action createAction( final long sleepMillis, final String name, Action... dependencies ) {
List<Action> deps = null;
if( dependencies != null ) {
deps = new ArrayList<>();
for( Action a : dependencies ) deps.add( a );
}
return Action.of( () -> {
System.out.println( "Service (" + name + ") started" );
try {
Thread.sleep( sleepMillis );
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println( "Service (" + name + ") completed" );
return true;
}, name, deps );
}
}
以及 Action 模型类。这代表一个操作和它所依赖的一系列操作。（与您最初的表示略有不同。但我认为，如果您相应地处理的话，无论哪种方式都可以。）
public class Action{
Callable<Boolean> action;
String name;
List<Action> dependencies = new ArrayList<>();
AtomicInteger status = new AtomicInteger( 0 ); //0 = Pending, 1 = Scheduled, 2 = Completed
public static final Object LOCK = new Object();
private Action(Callable<Boolean> action, String name, List<Action> dependencies) {
super();
this.action = action;
this.name = name;
if( dependencies != null ) this.dependencies = dependencies;
}
public static Action of( Callable<Boolean> action, String name, List<Action> dependencies ){
return new Action( action, name, dependencies );
}
public Callable<Boolean> getAction(){
return action;
}
public String getName(){
return name;
}
public List<Action> getDependencies(){
return dependencies;
}
public boolean isCompleted(){
return this.status.get() == 2;
}
public boolean isPending(){
return this.status.get() == 0;
}
public boolean isScheduled(){
return this.status.get() == 1;
}
public void setStatus( int status ){
this.status.getAndSet( status );
}
@Override
public int hashCode(){
final int prime = 31;
int result = 1;
result = prime * result + ((name == null) ? 0 : name.hashCode());
return result;
}
@Override
public boolean equals( Object obj ){
if (this == obj) return true;
if (obj == null) return false;
if (getClass() != obj.getClass()) return false;
Action other = (Action) obj;
if (name == null) {
if (other.name != null)
return false;
} else if (!name.equalsIgnoreCase( other.name )) return false;
return true;
}
}
| 归档时间： |
|
| 查看次数： |
386 次 |
| 最近记录： |