0

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

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

我想知道如何使用反应器总线请求/回复语法来替换这个同步服务代码。或者即使我应该使用完全不同的反应器构造。

这是遗留 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.runSync 调用其他服务的服务......我只是在寻找一个起点来研究如何可能使用反应器甚至另一个库来将这种类型的语法转换为异步非阻塞代码.

在这一点上，我正在考虑非常模糊的回调/某种 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-bus 文档，类似于以下内容，在很多层面上都没有意义，我只是对反应器了解得不够多，不知道为什么它没有意义，或者接下来要学习什么让我理解它的原因没有意义

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")); 
});

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

4

1 回答 1

0

您只需要使用 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.zip 或合并将所有值放在一起。但我相信这不是你需要的。

于 2016-03-03T09:05:32.730 回答