Apache Ignite 改装（一） -- 服务异步化支持

mgjj

本文假设读者了解Apache Ignite，阅读过ignite service grid的官方文档，或使用过ignite的service grid，本文同样假设读者了解 java的CompletionStage的相关用法。本文涉及的ignite版本为2.4.0。
　　使用Apache Ignite的Service grid作为微服务开发框架， 通常是如下定义和实现Service的：

服务接口：
　　

public interface MyService {　　public String sayHello(String to);
　　
}
　　

　　本文将实现如下样式的Service，使其异步化：

异步化的服务接口：
　　

public interface MyServiceAsync {　　public CompletionStage sayHello(String to);
　　
}
　　

　　当前ignite对上边这样的异步的service方法并没有remote支持。当调用端与服务部署再同一节点时，ignite会发起一个本地方法调用，这样是没有问题的，但是当服务部署端与调用端在不同节点时，ignite通过发起一个distributed task，将调用通过消息方式发布到服务部署节点，由于服务实现是异步的，通常来说，会返回一个未完成状态的CompletionStage，后续当真正complete的时候，调用端的CompletionStage并不会被notify，即调用端永远无法得到真正的调用结果。
　　为了能够支持CompletionStage的远程状态专递，我们需要对ignite进行如下改动：

org/apache/ignite/internal/processors/service/GridServiceProxy.java
　　

...　　
// line 192
　　
if(CompletionStage.class.isAssignableFrom(mtd.getReturnType())) {
　　//call async and notify completion stage
　　final IgniteInternalFuture future = ctx.closure().callAsyncNoFailover(
　　GridClosureCallMode.BROADCAST,
　　new ServiceProxyCallable(mtd.getName(), name, mtd.getParameterTypes(), args),
　　Collections.singleton(node),
　　false,
　　waitTimeout,
　　true);
　　CompletableFuture cs = new CompletableFuture() {
　　@Override
　　public boolean cancel(boolean mayInterruptIfRunning) {
　　try {
　　future.cancel();
　　} catch (Exception ignore) {}
　　return super.cancel(mayInterruptIfRunning);
　　}
　　};
　　future.listen(f -> {
　　if(f.error() != null) {
　　cs.completeExceptionally(f.error());
　　}else if(f.isCancelled()) {
　　cs.cancel(false);
　　}
　　if(f.isDone()) {
　　try {
　　Object result = f.get();
　　if(result != null && IgniteException.class.isAssignableFrom(result.getClass())) {
　　cs.completeExceptionally((IgniteException)result);
　　}else {
　　cs.complete(f.get());
　　}
　　} catch (IgniteCheckedException e) {
　　cs.completeExceptionally(e);
　　}
　　}
　　});
　　return cs;
　　
}
　　
...
　　

　　这段代码做了如下的事情：检测当服务方法返回值是一个CompletionStage的时候，则创建一个CompletableFuture作为代理对象返回给调用端。随后监听服务的远程调用的结果，并且用这个结果来更新这个CompletableFuture。到这里，调用端的service proxy的改造就完成了。接下来，我们还需要改造服务节点这一端：

org/apache/ignite/internal/processors/job/GridJobWorker.java（line 618起的这个finally块）,改造前：
　　

finally {　　// Finish here only if not held by this thread.
　　if (!HOLD.get())
　　finishJob(res, ex, sndRes);
　　else
　　// Make sure flag is not set for current thread.
　　// This may happen in case of nested internal task call with continuation.
　　HOLD.set(false);
　　ctx.job().currentTaskSession(null);
　　if (reqTopVer != null)
　　GridQueryProcessor.setRequestAffinityTopologyVersion(null);
　　
}
　　

改造后：
　　

finally {　　if(res != null && CompletionStage.class.isAssignableFrom(res.getClass())) {
　　final boolean sendResult = sndRes;
　　final IgniteException igException = ex;
　　@SuppressWarnings("unchecked")
　　CompletionStage cs = (CompletionStage)res;
　　cs.exceptionally(t->{
　　return new IgniteException(t);
　　}).thenAccept(r->{
　　if (!HOLD.get()) {
　　IgniteException e = igException;
　　finishJob(r, e, sendResult);
　　} else
　　// Make sure flag is not set for current thread.
　　// This may happen in case of nested internal task call with continuation.
　　HOLD.set(false);
　　ctx.job().currentTaskSession(null);
　　if (reqTopVer != null)
　　GridQueryProcessor.setRequestAffinityTopologyVersion(null);
　　});
　　} else {
　　// Finish here only if not held by this thread.
　　if (!HOLD.get())
　　finishJob(res, ex, sndRes);
　　else
　　// Make sure flag is not set for current thread.
　　// This may happen in case of nested internal task call with continuation.
　　HOLD.set(false);
　　ctx.job().currentTaskSession(null);
　　if (reqTopVer != null)
　　GridQueryProcessor.setRequestAffinityTopologyVersion(null);
　　}
　　
}
　　

　　这里做的事情是：当在服务部署节点上拿到执行结果的时候，如果发现服务返回结果是一个CompletionStage，那么处理这个CompletionStage的exceptionally和thenAccept， 把结果发送给remote的调用端。
　　就这样，通过简单的改装，我们使ignite有了处理异步服务方法调用的能力。下边我们实现一个服务来看看改装结果：

服务定义与实现：
　　

import java.util.concurrent.CompletionStage;　　
import org.apache.ignite.services.Service;
　　
public interface MyService extends Service {
　　public CompletionStage sayHelloAsync(String to);
　　public String sayHelloSync(String to);
　　
}
　　

import java.util.concurrent.CompletionStage;
　　
public>　　private ScheduledExecutorService es;
　　@Override public void init(ServiceContext ctx) throws Exception {
　　es = Executors.newSingleThreadScheduledExecutor();
　　}
　　@Override public CompletionStage sayHelloAsync(String to){
　　CompletableFuture ret = new CompletableFuture();
　　//return "async hello $to" after 3 secs
　　es.schedule(()->ret.complete("async hello " + to), 3, TimeUnit.SECONDS);
　　return ret;
　　}
　　@Override public String sayHelloSync(String to){
　　return "sync hello " + to;
　　}
　　...
　　
}
　　

　　然后将服务部署在Service grid中：
　　

...　　
ServiceConfiguration sConf = new ServiceConfiguration();
　　
sConf.setName("myservice.version.1");
　　
sConf.setService(new MyServiceImpl());
　　
sConf.setMaxPerNodeCount(2);
　　
sConf.setTotalCount(4);
　　
ignite.services().deploy(sConf);
　　
...
　　

　　然后启动一个客户端节点进行服务调用：
　　

MyService service = ignite.services().serviceProxy("myservice.version.1",  MyService.class, false);　　
//test async service
　　
service.sayHelloAsync("nathan").thenAccept(r->{
　　System.out.println(r);
　　
});
　　
//test sync service
　　
System.out.println(service.sayHelloSync("nathan"));
　　
...
　　

　　输出结果：
　　

sync hello nathan　　
async hello nathan
　　

　　可以看到先输出了sync的结果，大约3秒后输出async的结果。