【问题标题】:Is there an event-driven JSON REST client API for Java?是否有用于 Java 的事件驱动的 JSON REST 客户端 API?
【发布时间】:2014-01-03 11:38:31
【问题描述】:

我有一个 Java 应用程序,它使用 Spring 的 RestTemplate API 来编写简洁、可读的 JSON REST 服务使用者:

本质上:

 RestTemplate rest = new RestTemplate(clientHttpRequestFactory);
 ResponseEntity<ItemList> response = rest.exchange(url,
            HttpMethod.GET,     
            requestEntity,
            ItemList.class);

 for(Item item : response.getBody().getItems()) {
        handler.onItem(item);
 }

JSON 响应包含一个项目列表,如您所见,我在自己的代码中有一个事件驱动的设计来依次处理每个项目。但是,整个列表作为response 的一部分存在于内存中,由RestTemplate.exchange() 生成。

我希望应用程序能够处理包含大量项目的响应 - 比如说 50,000 个,在这种情况下,目前的实现存在两个问题:

  1. 在传输整个 HTTP 响应之前不会处理单个项目 - 增加了不必要的延迟。
  2. 巨大的响应对象位于内存中,在处理完最后一项之前无法进行 GC。

是否有相当成熟的 Java JSON/REST 客户端 API 以事件驱动的方式使用响应?

我想它会让你做这样的事情:

 RestStreamer rest = new RestStreamer(clientHttpRequestFactory);

 // Tell the RestStreamer "when, while parsing a response, you encounter a JSON
 // element matching JSONPath "$.items[*]" pass it to "handler" for processing.
 rest.onJsonPath("$.items[*]").handle(handler);

 // Tell the RestStreamer to make an HTTP request, parse it as a stream.
 // We expect "handler" to get passed an object each time the parser encounters
 // an item.
 rest.execute(url, HttpMethod.GET, requestEntity);

我很感激我可以使用来自 Jackson、GSON 等的流式 JSON API 来实现我自己的这种行为实现——但我很想被告知有一些东西可以通过简洁、富有表现力的 API 可靠地完成它,与 HTTP 方面集成。

【问题讨论】:

  • 这可能会有所帮助:stackoverflow.com/questions/444380/… 我倾向于先尝试 GSON。
  • 我进行了相当多的谷歌搜索,但找不到提供JsonParser 或类似名称的现成 Spring HttpMessageConverter。没有将其发布为答案,因为您明确提到这不是您要寻找的,但我认为滚动您自己的可能是最好的方法。恕我直言,它将成为 Spring 的一个很好的补充。我对 Jersey 也相当熟悉,但找不到任何提供对 JSON 流的事件驱动访问的东西。

标签: java json rest


【解决方案1】:

几个月后;回来回答我自己的问题。

我没有找到一个富有表现力的 API 来做我想做的事,但我能够通过将 HTTP 主体作为流获取并使用 Jackson JsonParser 来实现所需的行为:

  ClientHttpRequest request = 
        clientHttpRequestFactory.createRequest(uri, HttpMethod.GET);
  ClientHttpResponse response = request.execute();

  return handleJsonStream(response.getBody(), handler);

... 使用 handleJsonStream 设计来处理如下所示的 JSON:

 { items: [ 
      { field: value; ... }, 
      { field: value, ... },
      ... thousands more ... 
 ] }

...它验证导致数组开始的标记;每次遇到数组元素时,它都会创建一个Item 对象,并将其提供给处理程序。

 // important that the JsonFactory comes from an ObjectMapper, or it won't be
 // able to do readValueAs()
 static JsonFactory jsonFactory = new ObjectMapper().getFactory();

 public static int handleJsonStream(InputStream stream, ItemHandler handler) throws IOException {

     JsonParser parser = jsonFactory.createJsonParser(stream);

     verify(parser.nextToken(), START_OBJECT, parser);
     verify(parser.nextToken(), FIELD_NAME, parser);
     verify(parser.getCurrentName(), "items", parser);
     verify(parser.nextToken(), START_ARRAY, parser);
     int count = 0;
     while(parser.nextToken() != END_ARRAY) {
        verify(parser.getCurrentToken(), START_OBJECT, parser);
        Item item = parser.readValueAs(Item.class);
        handler.onItem(item);
        count++;
     }
     parser.close(); // hope it's OK to ignore remaining closing tokens.
     return count;
 }

verify() 只是一个私有静态方法,如果前两个参数不相等,则会引发异常。

这个方法的关键在于,不管流中有多少个item,这个方法只有一个item的引用。

【讨论】:

    【解决方案2】:

    您可以尝试JsonSurfer,它旨在以事件驱动的方式处理 json 流。

    JsonSurfer surfer = JsonSurfer.jackson();
    Builder builder = config();
    builder.bind("$.items[*]", new JsonPathListener() {
            @Override
            public void onValue(Object value, ParsingContext context) throws Exception {
                // handle the value
            }
        });
    surfer.surf(new InputStreamReader(response.getBody()), builder.build());
    

    【讨论】:

    • 这正是我想要的。现在为时已晚,我使用了自己的答案中描述的解决方案。但是假设 JsonSurfer 是稳定的,它回答了这个问题,所以我已经接受了。
    【解决方案3】:

    有没有办法分解请求?听起来您应该使用分页。使其可以请求前 100 个结果,接下来的 100 个结果,依此类推。该请求应采用起始索引和计数。这对于 REST 服务来说是非常常见的行为,听起来像是解决您的问题的方法。

    REST 的全部意义在于它是无状态的,听起来您正试图使其成为有状态的。这对 REST 来说是一种诅咒,因此您不会找到任何以这种方式编写的库。

    REST 的事务性质在设计上是有意为之的,因此您不会轻易解决这个问题。如果您尝试,您将与谷物作斗争。

    【讨论】:

    • 我认为所寻求的并不违反资源的 RESTful/无状态性质。无论对每个项目采取何种操作,在发出的响应中表示仍然是“无状态的”。
    • 正如@cmonkey 所说,我看不出“大”不应该是“无状态”的任何理由。我只想消耗一个长响应,而不是一次将其全部保存在内存中。当底层列表可能从一个请求更改为下一个请求时(例如,它是一个包含正在添加/删除的项目的邮箱),分页很容易出错。
    【解决方案4】:

    据我所见,包装框架(就像您正在使用的那样)通过将响应反序列化为对象使事情变得容易。在您的情况下,是一组对象。

    但是,要以流方式使用事物,您可能需要获取底层 HTTP 响应流。我最熟悉的Jersey,暴露了https://jersey.java.net/nonav/apidocs/1.5/jersey/com/sun/jersey/api/client/ClientResponse.html#getEntityInputStream()

    它将通过调用来使用

    Client client = Client.create();
    WebResource webResource = client.resource("http://...");
    ClientResponse response = webResource.accept("application/json")
                   .get(ClientResponse.class);
    InputStream is = response.getEntityInputStream();
    

    这为您提供了传入的数据流。下一步是编写流部分。鉴于您使用的是 JSON,有不同级别的选项,包括 http://wiki.fasterxml.com/JacksonStreamingApihttp://argo.sourceforge.net/documentation.html。他们可以使用 InputStream。

    这些并没有真正充分利用可以完成的完整反序列化,但您可以使用它们来解析 json 数组的元素,并将该项目传递给典型的 JSON 对象映射器,(如 Jackson, GSON 等)。这成为事件处理逻辑。您可以为此生成新线程,或执行您的用例需要的任何操作。

    【讨论】:

    • 这就是我所说的“我很感激我可以使用来自 Jackson、GSON 等的流式 JSON API 来实现我自己的这种行为。” -- 但是没有其他建议似乎意味着它是唯一可用的路线,所以谢谢。
    • ngdata.com/parsing-a-large-json-file-efficiently-and-easily,你可以在Jackson中混合流式和树解析。每个数组成员调用的方法是 jp.readValueAsTree();
    • 这是导致解决方案的答案,但我接受了自己的答案,因为我已经给出了完整的源示例。那谢谢啦。感谢@flup 指出混合流/映射。
    【解决方案5】:

    我不会声称知道所有其他框架(甚至一半),但我会回答这个问题

    可能不会

    正如其他人所指出的,这不是 REST 通常认为的交互方式。 REST 是一个很棒的 Hammer,但是如果您需要流式传输,那么您(恕我直言)处于螺丝刀领域,并且该锤子可能仍然可以工作,但它可能会造成混乱。人们可以整天争论它是否与 REST 一致,但最终我会非常惊讶地发现一个实现此功能的框架。如果该功能成熟(即使框架成熟),我会更加惊讶,因为对于 REST,您的用例充其量只是一个不常见的极端情况。

    如果有人确实想出了一个,我会很乐意纠正并学习新的东西:)

    对于这个特定的操作,也许最好考虑彗星或网络套接字。 This question 可能会有所帮助,因为您已经有了春天。 (如果您需要支持 IE not really viable,大多数商业应用程序仍然需要它......遗憾的是,在我的个人工作中,我有一个客户和一个关键客户仍在使用 IE 7)

    【讨论】:

      【解决方案6】:

      你可以考虑Restlet

      http://restlet.org/discover/features

      支持异步请求处理,与IO操作解耦。与 Servlet API 不同,Restlet 应用程序没有直接控制输出流,它们只提供由服务器连接器写入的输出表示。

      【讨论】:

      • 感谢您查看问题 - 但您的块引用中提到的 Restlet 功能都没有解决问题中描述的需求。我快速浏览了 Restlet 文档,它似乎可以一次性将 HTTP 内容反序列化为一个对象,就像 Spring RestTemplate 一样。它不公开 InputStream,因此您甚至无法滚动自己的事件驱动解析。
      【解决方案7】:

      实现这一点的最佳方法是使用另一个用于 JVM 的流式运行时,它允许从 websocket 读取响应,我知道有一个叫做 atmostsphere

      通过这种方式,您的大型数据集在两侧以块的形式发送和接收,并以相同的方式实时读取,而无需等待整个响应。

      这有一个很好的 POC: http://keaplogik.blogspot.in/2012/05/atmosphere-websockets-comet-with-spring.html

      服务器:

          @RequestMapping(value="/twitter/concurrency")
      @ResponseBody
      public void twitterAsync(AtmosphereResource atmosphereResource){
          final ObjectMapper mapper = new ObjectMapper();
      
          this.suspend(atmosphereResource);
      
          final Broadcaster bc = atmosphereResource.getBroadcaster();
      
          logger.info("Atmo Resource Size: " + bc.getAtmosphereResources().size());
      
          bc.scheduleFixedBroadcast(new Callable<String>() {
      
              //@Override
              public String call() throws Exception {
      
                  //Auth using keaplogik application springMVC-atmosphere-comet-webso key
                  final TwitterTemplate twitterTemplate = 
                      new TwitterTemplate("WnLeyhTMjysXbNUd7DLcg",
                              "BhtMjwcDi8noxMc6zWSTtzPqq8AFV170fn9ivNGrc", 
                              "537308114-5ByNH4nsTqejcg5b2HNeyuBb3khaQLeNnKDgl8",
                              "7aRrt3MUrnARVvypaSn3ZOKbRhJ5SiFoneahEp2SE");
      
                  final SearchParameters parameters = new SearchParameters("world").count(5).sinceId(sinceId).maxId(0);
                  final SearchResults results = twitterTemplate.searchOperations().search(parameters);
      
                  sinceId = results.getSearchMetadata().getMax_id();
      
                  List<TwitterMessage> twitterMessages = new ArrayList<TwitterMessage>();
      
                  for (Tweet tweet : results.getTweets()) {
                      twitterMessages.add(new TwitterMessage(tweet.getId(),
                                                             tweet.getCreatedAt(),
                                                             tweet.getText(),
                                                             tweet.getFromUser(),
                                                             tweet.getProfileImageUrl()));
                  }
      
                  return mapper.writeValueAsString(twitterMessages);
              }
      
          }, 10, TimeUnit.SECONDS);
      }
      

      客户: Atmosphere 有自己的 javascript 文件来处理不同的 Comet/Websocket 传输类型和请求。通过使用它,您可以将 Spring URL Controller 方法端点设置为请求。订阅控制器后,您将收到分派,可以通过添加 request.onMessage 方法来处理。这是一个带有 websocket 传输的示例请求。

             var request = new $.atmosphere.AtmosphereRequest();
         request.transport = 'websocket';
         request.url = "<c:url value='/twitter/concurrency'/>";
         request.contentType = "application/json";
         request.fallbackTransport = 'streaming';
      
         request.onMessage = function(response){
             buildTemplate(response);
         };
      
         var subSocket = socket.subscribe(request);
      
         function buildTemplate(response){
      
           if(response.state = "messageReceived"){
      
                var data = response.responseBody;
      
              if (data) {
      
                  try {
                      var result =  $.parseJSON(data);
      
                      $( "#template" ).tmpl( result ).hide().prependTo( "#twitterMessages").fadeIn();
      
                  } catch (error) {
                      console.log("An error ocurred: " + error);
                  }
              } else {
                  console.log("response.responseBody is null - ignoring.");
              }
          }
         }
      

      它支持所有主流浏览器和本地移动客户端 Apple 是这项技术的先驱:

      正如这里提到的,对 Web 和企业 JEE 容器上的部署环境的出色支持:

      http://jfarcand.wordpress.com/2012/04/19/websockets-or-comet-or-both-whats-supported-in-the-java-ee-land/

      【讨论】:

        猜你喜欢
        • 2012-08-22
        • 2012-07-29
        • 1970-01-01
        • 1970-01-01
        • 2019-10-04
        • 1970-01-01
        • 2015-12-24
        • 2011-12-31
        • 1970-01-01
        相关资源
        最近更新 更多