我应该使用什么消息模式来处理和返回来自休息请求的响应？答案

【问题标题】：What messaging pattern should I use to process and return responses from a rest request?我应该使用什么消息模式来处理和返回来自休息请求的响应？
【发布时间】：2016-08-08 21:46:51
【问题描述】：

我有一个类似这样的中心辐射型架构：

GET 请求进入集线器并将其路由到其中一个辐条进行处理。在集线器上，我还将请求放在带有 UUID 的地图中，这样当我从处理中取回数据时，我可以返回正确的响应。辐条是相同的，用于平衡负载。然后，我需要将信息从辐条传递回集线器并返回正确的响应。
我想使用 JMS 进行消息传递。

integration patterns 的最佳组合是什么？

【问题讨论】：

标签： java apache-camel jms vert.x

【解决方案1】：

您已经在 Vert.x 中有请求/回复，因此您可以使用大约 20 行代码来实现此行为：

public static void main(String[] args) {

    Vertx vertx = Vertx.vertx();

    Router router = Router.router(vertx);

    router.get("/").handler((request) -> {
        // When hub receives request, it dispatches it to one of the Spokes
        String requestUUID = UUID.randomUUID().toString();
        vertx.eventBus().send("processMessage", requestUUID, (spokeResponse) -> {
            if (spokeResponse.succeeded()) {
                request.response().end("Request " + requestUUID + ":" + spokeResponse.result().body().toString());
            }
            // Handle errors
        });
    });

    // We create two Spokes
    vertx.deployVerticle(new SpokeVerticle());
    vertx.deployVerticle(new SpokeVerticle());

    // This is your Hub
    vertx.createHttpServer().requestHandler(router::accept).listen(8888);
}

这是 Spoke 的样子：

/**
 * Static only for the sake of example
 */
static class SpokeVerticle extends AbstractVerticle {

    private String id;

    @Override
    public void start() {
        this.id = UUID.randomUUID().toString();

        vertx.eventBus().consumer("processMessage", (request) -> {
            // Do something smart

            // Reply
            request.reply("I'm Spoke " + id + " and my reply is 42");
        });
    }
}

尝试在浏览器中访问它http://localhost:8888/

您应该会看到每次都会生成请求 ID，而两个 Spoke 中只有一个回答您的请求。

【讨论】：

谢谢，这正是我想要的。

【解决方案2】：

好吧，如果我正确理解您的设计，这似乎是请求/回复场景，因为辐条实际上正在返回一些响应。如果不是，它将是发布/订阅。

您可以将 ActiveMQ 用于 jms 和请求/回复。看这里： http://activemq.apache.org/how-should-i-implement-request-response-with-jms.html

至于细节，这完全取决于您的要求，响应会立即发送还是需要长时间运行？

如果这是一个长时间运行的过程，您可以避免请求/回复并使用触发后忘记的场景。

基本上，集线器会在一个队列中触发一条消息，该队列正被其中一个分支组件侦听。后端处理完成后，它会将响应返回到集线器监控的队列。您可以通过一些相关 ID 关联请求/响应。在请求部分，您可以将correlationId 保存在缓存中以匹配响应。在请求/回复场景中，这是由基础架构为您完成的，但不用于长时间运行的流程。

总结一下：使用 ActiveMQ 通过 JMS 进行消息处理。使用 Camel 作为 REST 位。如果您确定您期望得到相当快的响应，请使用请求/回复。如果您预计响应需要很长时间但必须匹配消息的correlationIds，请使用fire and forget。

【讨论】：

【解决方案3】：

如果您希望将 Camel 与 JMS 一起使用，那么您应该使用 Request-Reply EIP，就示例而言，您有一个 pretty good one provided via Camel's official examples - 它可能有点旧，但仍然非常有效。

虽然您可以通过 Spring 忽略示例的 Camel 配置，但它的路由定义提供了足够的信息：

public class SpokeRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        from("jms:queue:spoke")
            .process(e -> {
               Object result = ...; // do some processing
               e.getIn().setBody(result); // publish the result
        // Camel will automatically reply if it finds a ReplyTo and CorrelationId headers
            });
    }
}

那么HUB需要做的就是调用：

ProducerTemplate camelTemplate = camelContext.createProducerTemplate();
Object response = camelTemplate.sendBody("jms:queue:spoke", ExchangePattern.InOut, input);

【讨论】：