Vertx 抛出 IllegalStateException：响应已经写入

Question

我正在构建一个 rest api，它从其他人那里收集数据 api，用它做一些逻辑并发回客户端：

我的主要class:

public class Main {

public static void main(String[] args) {

    Vertx.vertx().deployVerticle(RestVerticle.class.getName());
}

这是我的 RestVerticle：

public class RestVerticle extends AbstractVerticle {

public static final String API_V1 = "/api/v1";

private Map<String, JsonObject> products = new HashMap<>();

@Override
public void start() {

    Router router = Router.router(vertx);

    router.route().handler(BodyHandler.create());
    router.get(API_V1 + "/business/all").handler(this::getAllBusinesses);

    vertx.createHttpServer().requestHandler(router::accept).listen(8080);
}

private void getAllBusinesses(RoutingContext routingContext) {

    vertx.deployVerticle(YelpClientVerticle.class.getName());
    MessageConsumer<String> consumer = vertx.eventBus().consumer("api");

    consumer.handler(message -> {
        JsonObject m = new JsonObject((String) message.body());
        System.out.println("Received message: " + message.body());
        routingContext.response().putHeader("content-type", "application/json").end(m.encodePrettily());

    });
}

}

这是我的 httpclient ，它调用 Yelp API:

public class YelpClientVerticle extends AbstractVerticle {

private static final String API_HOST = "api.yelp.com";
private static final String DEFAULT_TERM = "dinner";
private static final String DEFAULT_LOCATION = "San Francisco, CA";
private static final int SEARCH_LIMIT = 20;
private static final String SEARCH_PATH = "/v2/search";
private static final String BUSINESS_PATH = "/v2/business";

/*
 * Update OAuth credentials below from the Yelp Developers API site:
 * http://www.yelp.com/developers/getting_started/api_access
 */
private static final String CONSUMER_KEY = "XXXXX";
private static final String CONSUMER_SECRET = "XXXXX";
private static final String TOKEN = "XXXXX";
private static final String TOKEN_SECRET = "XXXXX";

OAuthService service;
Token accessToken;

/**
 * Setup the Yelp API OAuth credentials.
 *
 * @param consumerKey    Consumer key
 * @param consumerSecret Consumer secret
 * @param token          Token
 * @param tokenSecret    Token secret
 */

/**
 * Creates and sends a request to the Search API by term and location.
 * <p>
 * See <a href="http://www.yelp.com/developers/documentation/v2/search_api">Yelp Search API V2</a>
 * for more info.
 *
 * @param term     <tt>String</tt> of the search term to be queried
 * @param location <tt>String</tt> of the location
 * @return <tt>String</tt> JSON Response
 */
public String searchForBusinessesByLocation(String term, String location) {
    OAuthRequest request = createOAuthRequest(SEARCH_PATH);
    request.addQuerystringParameter("term", term);
    request.addQuerystringParameter("location", location);
    request.addQuerystringParameter("limit", String.valueOf(SEARCH_LIMIT));
    return sendRequestAndGetResponse(request);
}

/**
 * Creates and sends a request to the Business API by business ID.
 * <p>
 * See <a href="http://www.yelp.com/developers/documentation/v2/business">Yelp Business API V2</a>
 * for more info.
 *
 * @param businessID <tt>String</tt> business ID of the requested business
 * @return <tt>String</tt> JSON Response
 */
public String searchByBusinessId(String businessID) {
    OAuthRequest request = createOAuthRequest(BUSINESS_PATH + "/" + businessID);
    return sendRequestAndGetResponse(request);
}

/**
 * Creates and returns an {@link OAuthRequest} based on the API endpoint specified.
 *
 * @param path API endpoint to be queried
 * @return <tt>OAuthRequest</tt>
 */
private OAuthRequest createOAuthRequest(String path) {
    OAuthRequest request = new OAuthRequest(Verb.GET, "https://" + API_HOST + path);
    return request;
}

/**
 * Sends an {@link OAuthRequest} and returns the {@link Response} body.
 *
 * @param request {@link OAuthRequest} corresponding to the API request
 * @return <tt>String</tt> body of API response
 */
private String sendRequestAndGetResponse(OAuthRequest request) {
    System.out.println("Querying " + request.getCompleteUrl() + " ...");
    this.service.signRequest(this.accessToken, request);
    Response response = request.send();
    return response.getBody();
}

/**
 * Queries the Search API based on the command line arguments and takes the first result to query
 * the Business API.
 *
 * @param yelpApiCli <tt>YelpAPICLI</tt> command line arguments
 */
private String queryAPI(YelpAPICLI yelpApiCli) {
    String searchResponseJSON =
            searchForBusinessesByLocation(yelpApiCli.term, yelpApiCli.location);

    JSONParser parser = new JSONParser();
    JSONObject response = null;
    try {
        response = (JSONObject) parser.parse(searchResponseJSON);
    } catch (ParseException pe) {
        System.out.println("Error: could not parse JSON response:");
        System.out.println(searchResponseJSON);
        System.exit(1);
    }

    JSONArray businesses = (JSONArray) response.get("businesses");
    JSONObject firstBusiness = (JSONObject) businesses.get(0);
    String firstBusinessID = firstBusiness.get("id").toString();
    System.out.println(String.format(
            "%s businesses found, querying business info for the top result \"%s\" ...",
            businesses.size(), firstBusinessID));

    // Select the first business and display business details
    String businessResponseJSON = searchByBusinessId(firstBusinessID.toString());
    System.out.println(String.format("Result for business \"%s\" found:", firstBusinessID));
    System.out.println(businessResponseJSON);

    return businessResponseJSON;
}

/**
 * Command-line interface for the sample Yelp API runner.
 */
private static class YelpAPICLI {
    @Parameter(names = {"-q", "--term"}, description = "Search Query Term")
    public String term = DEFAULT_TERM;

    @Parameter(names = {"-l", "--location"}, description = "Location to be Queried")
    public String location = DEFAULT_LOCATION;
}

@Override
public void start() throws Exception {
    // Note! in real-life you wouldn't often set trust all to true as it could leave you open to man in the middle attacks.
    this.service =
            new ServiceBuilder().provider(TwoStepOAuth.class).apiKey(CONSUMER_KEY)
                    .apiSecret(CONSUMER_SECRET).build();
    this.accessToken = new Token(TOKEN, TOKEN_SECRET);
    YelpAPICLI yelpApiCli = new YelpAPICLI();
    new JCommander(yelpApiCli);


    String response = queryAPI(yelpApiCli);
    vertx.eventBus().send("api", response);
}

}

我运行有 2 个问题。

第一个问题是 Yelp 客户端处理请求的时间太长，它阻塞了主线程并发出警告：

Jan 04, 2016 1:34:30 AM io.vertx.core.impl.BlockedThreadChecker
WARNING: Thread Thread[vert.x-eventloop-thread-4,5,main] has been blocked for 3151 ms, time limit is 2000

第二个问题是，在完成第一个请求后，例如，我第一次去我的localhost:8080/api/v1/business/all，请求return成功，但是下次再次点击 URL，它会抛出这样的异常：

Jan 04, 2016 1:34:30 AM io.vertx.core.eventbus.impl.HandlerRegistration
SEVERE: Failed to handleMessage
java.lang.IllegalStateException: Response has already been written

如何解决这两个问题？

Answer 1

问题是您正在做所有工作 - 启动 Yelp Verticle，并在事件总线上注册消费者 - 在每个请求上。这不可能是你想要的。

所以，我认为正在发生的事情：

您提出休息要求API。
执行getAllBusinesses()方法。
YelpClientVerticle 已启动。
处理程序已在 api 端点上注册，该处理程序将写入响应。
您的 YelpClientVerticle 做了很多阻塞工作 - 这就是您收到 BlockedThreadChecker 警告的原因。
最后，Yelp 请求 returns 并通过事件总线发送消息，然后写入响应。
您再次请求
转到 2

您的问题是，对于每个请求，您都会启动 另一个 YelpClientVerticle，并注册另一个处理程序来监听 相同的[=65] =] EventBus端点地址。

让多个处理程序监听同一个 EventBus address 是完全可以接受的。当发生这种情况时，Vert.x 以 RoundRobin 的方式选择其中一个处理程序。

我猜测在 第二个请求 上，Vertx 正在选择 第一个处理程序 ，并尝试写入到您已经写入的第一个请求的响应。因此错误。

我会尝试将 YelpClientVerticle 的部署移动到 RestVerticle 的启动中 - 那么您将只有一个实例。

您可能想调换发件人/消费者，因此您 send 向 YelpClientVerticle 发送消息，后者随后回复。

您可能还想阅读有关 Running Blocking Code 的文档，因为您的 Yelp 客户端看起来正在阻塞。

希望对您有所帮助

Answer 2

此答案特定于异常 "response has already been written"。有可能在我们的代码中使用 "routingContext.next()" 和 "routingContext.response.end()"。如果没有可用于路由的处理程序，则 vert.x 将抛出上述异常。

Vertx 抛出 IllegalStateException：响应已经写入

Vertx throws IllegalStateException : Response has already been written

java

asynchronous

vert.x