重构旧版SOA系统以使用非阻塞异步微服务

Question

我刚刚开始尝试了解RxJava，以便我可以使用项目反应器重构旧版SOA系统来使用非阻塞异步微服务。

目前，我正在做一项可行性研究，并考虑使用类似spoon的东西来转换遗留服务代码（但这与这个问题无关）

我想知道如何使用reactor bus Request/Reply语法来替换这个同步服务代码。或者即使我应该使用完全不同的Reactor结构。

这里是一个遗留soa服务的示例，它是人为设计的，因此可能没有完美的意义，但基本上每个服务都依赖于最后一个服务的结果。

 public static Map<String, Object> createAccount(DispatchContext dctx, Map<String, Object> context) {
    LocalDispatcher dispatcher = dctx.getDispatcher();

    String accountPartyId = (String) context.get("partyId");

    Map<String, Object> input = UtilMisc.toMap("groupName", context.get("accountName"), "groupNameLocal", context.get("groupNameLocal"), "officeSiteName", context.get("officeSiteName"), "description", context.get("description"), "partyId", accountPartyId);

    Map<String, Object> serviceResults1 = dispatcher.runSync("createPartyGroup", input);

    Map<String, Object> serviceResults2 = dispatcher.runSync("createPartyRole", UtilMisc.toMap("partyId", (String) serviceResults1.get("partyId"), "roleTypeId", "ACCOUNT"));

    String dataSourceId = (String) context.get("dataSourceId");
    Map<String, Object> serviceResults3 = null;
    if (dataSourceId != null) {
        serviceResults3 = dispatcher.runSync("crmsfa.addAccountDataSource", UtilMisc.toMap("partyId", (String) serviceResults2.get("partyId"), "dataSourceId", dataSourceId));
    }

    String marketingCampaignId = (String) context.get("marketingCampaignId");
    Map<String, Object> serviceResults4 = null;
    if (marketingCampaignId != null) {
        serviceResults4 = dispatcher.runSync("crmsfa.addAccountMarketingCampaign", UtilMisc.toMap("partyId", (String) serviceResults3.get("partyId"), "marketingCampaignId", marketingCampaignId));

    }

    String initialTeamPartyId = (String) context.get("initialTeamPartyId");
    Map<String, Object> serviceResults5 = null;
    if (initialTeamPartyId != null) {
        serviceResults5 = dispatcher.runSync("crmsfa.assignTeamToAccount", UtilMisc.toMap("accountPartyId", (String) serviceResults4.get("partyId"), "teamPartyId", initialTeamPartyId, "userLogin", userLogin));
    }

    Map<String, Object> results = ServiceUtil.returnSuccess();
    results.put("groupId", (String) serviceResults1.get("groupId"));
    results.put("roleId", (String) serviceResults2.get("roleId"));
    results.put("dataSourceId", (String) serviceResults3.get("dataSourceId"));
    results.put("marketingCampaignId", (String) serviceResults4.get("marketingCampaignId"));
    results.put("teamPartyId", (String) serviceResults5.get("teamPartyId"));
    return results;
}

基本上，这是一个使用dispatcher调用其他服务的服务。运行同步。。。我只是在寻找一个起点，研究如何可能使用reactor或其他库将这种语法转换为异步非阻塞代码。

在这一点上，我用非常模糊的术语来思考回调/某种promise类型的结构。

就像第一次呼叫另一个服务是

Map<String, Object> serviceResults = dispatcher.runSync("createPartyGroup", input);

如果它返回了一个包含serviceResults映射的Promise对象，那么该方法的其余部分可以被移动到Promise onComplete块中，结果将是构成该服务方法的一组深度嵌套的onComplete代码块。

Promise p = task {
    // createPartyGroup service call
}
p.onComplete { result ->

Promise p2 = task {
    // createPartyRole sevice call
}

p2.onComplete { result ->
//next service call
}   
}  
}

或者看看Reactor总线文档，比如下面的内容，在很多层面上都没有意义，我只是对Reactor的了解不够，不知道为什么它没有意义，或者我接下来要学习什么来理解为什么它没有意义

bus.send("service.createPartyGroup", Event.wrap(input, "reply.service.createPartyGroup")); 
bus.receive($("reply.service.createPartyGroup"), ev -> {
Map<?> input2 = UtilMisc.toMap("partyId", (String) ev.get("partyId"), "roleTypeId", "ACCOUNT")
  bus.send("service.createPartyRole", Event.wrap(input2, "reply.service.createPartyRole")); 
}); 

我意识到开始研究反应式编程范式是一个相当奇怪的地方。但是，替换这个同步服务代码是我的最终目标，如果我至少理解了语法，我就可以从中逆向工作。

Answer 1

您只需要使用Observable，在您的流中，一个已发射的项目通过流。检查文档https://github.com/ReactiveX/RxJava

这将是一个连续的流程

  Observable.just(methodThatCallFirstServiceAndReturnObservable(params))
            .flatMap(resul1 -> methodThatCallSecondAndReturnObservable(resul1))
            .flatMap(resul2 -> methodThatCallThirdAndReturnObservable(resul2))
            .subscribe(result3->"Last value emmited here:");

您可以并行运行这三个服务调用，并使用Observable将所有值汇总在一起。压缩或合并。但我相信这不是你需要的。