大朗仿做网站,杭州笕桥网站建设,江汉区建设局官方网站,海南省最新消息本文将涉及多个主题#xff1a;响应式编程#xff0c;HTTP#xff0c;解析JSON以及与社交API集成。 完全在一个用例中#xff1a;我们将通过非夸张的RxNetty库实时加载和处理新的metup.com事件#xff0c;结合Netty框架的强大功能和RxJava库的灵活性。 Meetup提供了公开可… 本文将涉及多个主题响应式编程HTTP解析JSON以及与社交API集成。 完全在一个用例中我们将通过非夸张的RxNetty库实时加载和处理新的metup.com事件结合Netty框架的强大功能和RxJava库的灵活性。 Meetup提供了公开可用的流API 可实时推送世界各地注册的每一个Meetup。 只需浏览至stream.meetup.com/2/open_events并观察JSON块如何缓慢地出现在屏幕上。 每当有人创建新事件时自包含的JSON就会从服务器推送到您的浏览器。 这意味着这样的请求永无止境相反只要需要我们就会不断接收部分数据。 我们已经在将Twitter4J变成RxJava的Observable中研究了类似的情况。 每个新的Meetup事件都会发布一个独立的JSON文档与此类似省略很多细节 { id : 219088449,name : Silver Wings Brunch,time : 1421609400000,mtime : 1417814004321,duration : 900000,rsvp_limit : 0,status : upcoming,event_url : http://www.meetup.com/Laguna-Niguel-Social-Networking-Meetup/events/219088449/,group : { name : Former Flight Attendants South Orange and North San Diego Co,state : CA...},venue : { address_1 : 26860 Ortega Highway,city : San Juan Capistrano,country : US...},venue_visibility : public,visibility : public,yes_rsvp_count : 1...
} 每当我们长时间轮询的HTTP连接带有Transfer-Encoding: chunked响应标头推送此类JSON时我们都希望对其进行解析并以某种方式进一步传递。 我们讨厌回调因此RxJava似乎是一个合理的选择认为 ObservableEvent 。 步骤1使用RxNetty接收原始数据 我们不能使用普通的HTTP客户端因为它们专注于请求-响应语义。 这里没有任何响应我们只是永远保持打开的连接并在数据到达时使用它们。 RxJava具有开箱即用的RxApacheHttp库但它假定为text/event-stream内容类型 。 相反我们将使用底层的通用RxNetty库。 它是Nettyduh的包装并且能够实现任意的 TCP / IP包括HTTP以及UDP客户端和服务器。 如果您不了解Netty则它是基于数据包的而不是面向流的因此我们可以预期每次Meetup推送都会有一个Netty事件。 该API当然不是简单明了的但是一旦您使用它它就会变得有意义 HttpClientByteBuf, ByteBuf httpClient RxNetty.ByteBuf, ByteBufnewHttpClientBuilder(stream.meetup.com, 443).pipelineConfigurator(new HttpClientPipelineConfigurator()).withSslEngineFactory(DefaultFactories.trustAll()).build();final ObservableHttpClientResponse responses httpClient.submit(HttpClientRequest.createGet(/2/open_events));
final Observable byteBufs responses.flatMap(AbstractHttpContentHolder::getContent);
final Observable chunks byteBufs.map(content - content.toString(StandardCharsets.UTF_8)); 首先我们创建HttpClient并设置SSL请注意关于服务器证书的trustAll()可能不是最佳的生产设置。 稍后我们submit() GET请求并接收ObservableHttpClientResponseByteBuf作为回报。 ByteBuf是Netty对通过网络发送或接收的一堆字节的抽象。 该观察结果将立即告诉我们从Meetup收到的每条数据。 从响应中提取ByteBuf 我们将其转换为包含上述JSON的String 。 到目前为止一切正常。 步骤2将数据包与JSON文档对齐 Netty非常强大因为它不会掩盖泄漏抽象所固有的复杂性。 每次通过TCP / IP线路接收到某些内容时都会通知我们。 您可能会相信当服务器发送100字节时客户端的Netty会通知我们有关这100字节的信息。 但是TCP / IP堆栈可以自由地拆分和合并您通过有线发送的数据特别是因为它假定是流因此如何将其拆分为数据包无关紧要。 Netty的文档中对此警告做了很大的解释。 对我们意味着什么 当Meetup发送单个事件时我们可能仅收到一个可观察到的chunks String 。 但是同样可以将其划分为任意数量的数据包因此chunks将发出多个String 。 更糟糕的是如果Meetup接连发送两个事件则它们可能适合一个数据包。 在这种情况下 chunks将发出一个带有两个独立JSON文档的String 。 事实上我们不能假设JSON字符串和收到的网络数据包之间有任何对齐。 我们所知道的是代表事件的各个JSON文档由换行符分隔。 令人惊讶的是 RxJavaString官方附加组件RxJavaString提供了一种精确的方法 Observable jsonChunks StringObservable.split(chunks, \n); 实际上甚至还有更简单的StringObservable.byLine(chunks) 但它使用的是平台相关的行尾。 最好在官方文档中解释split()作用 现在我们可以安全地解析jsonChunks发出的每个String了 步骤3解析JSON 有趣的是这一步骤并不是那么简单。 我承认我排序的享受WSDL时间因为我很容易可预见生成如下web服务的合同Java模型。 JSON特别是在JSON模式的边缘市场渗透方面基本上是集成的“狂野西部”。 通常您会得到非正式的文档或请求和响应的样本。 没有类型信息或格式无论字段是否为必填项等等。此外由于我不情愿使用地图映射 在那里Clojure程序员为了使用基于JSON的REST服务我必须自己编写映射POJO。 好吧有解决方法。 首先我举了一个由Meetup流API生成的JSON的代表性示例并将其放在src/main/json/meetup/event.json 。 然后我使用jsonschema2pojo-maven-plugin 也存在Gradle和Ant版本。 插件的名称令人困惑它还可以与JSON示例不仅是架构一起使用以生成Java模型 plugingroupIdorg.jsonschema2pojo/groupIdartifactIdjsonschema2pojo-maven-plugin/artifactIdversion0.4.7/versionconfigurationsourceDirectory${basedir}/src/main/json/meetup/sourceDirectorytargetPackagecom.nurkiewicz.meetup.generated/targetPackageincludeHashcodeAndEqualstrue/includeHashcodeAndEqualsincludeToStringtrue/includeToStringinitializeCollectionstrue/initializeCollectionssourceTypeJSON/sourceTypeuseCommonsLang3true/useCommonsLang3useJodaDatestrue/useJodaDatesuseLongIntegerstrue/useLongIntegersoutputDirectorytarget/generated-sources/outputDirectory/configurationexecutionsexecutionidgenerate-sources/idphasegenerate-sources/phasegoalsgoalgenerate/goal/goals/execution/executions
/plugin 此时Maven将创建与Jackson兼容的Event.java Venue.java Group.java等 private Event parseEventJson(String jsonStr) {try {return objectMapper.readValue(jsonStr, Event.class);} catch (IOException e) {throw new UncheckedIOException(e);}
} 很好它很好 final Observableevents jsonChunks.map(this::parseEventJson); 步骤5获利 有了ObservableEvent我们可以实现一些非常有趣的用例。 是否要查找刚刚创建的波兰所有聚会的名称 当然 events.filter(event - event.getVenue() ! null).filter(event - event.getVenue().getCountry().equals(pl)).map(Event::getName).forEach(System.out::println); 寻找统计信息每分钟创建多少个事件 没问题 events.buffer(1, TimeUnit.MINUTES).map(List::size).forEach(count - log.info(Count: {}, count)); 或者也许您想继续搜索将来最远的聚会而不是寻找比已发现的聚会更近的聚会 events.filter(event - event.getTime() ! null).scan(this::laterEventFrom).distinct().map(Event::getTime).map(Instant::ofEpochMilli).forEach(System.out::println);//...private Event laterEventFrom(Event first, Event second) {return first.getTime() second.getTime() ?first :second;
} 此代码过滤掉未知时间的事件发出当前事件或前一个事件 scan() 具体取决于后面的事件过滤出重复事件并显示时间。 这个运行了几分钟的小程序已经发现一个计划于2015年11月创建的聚会而在撰写本文时聚会是2014年12月。 可能性是无止境的。 希望我能对如何轻松地将各种技术融合在一起有一个很好的了解反应式编程以编写超快速的网络代码无需样板代码的类型安全JSON解析以及RxJava来快速处理事件流。 请享用 翻译自: https://www.javacodegeeks.com/2014/12/accessing-meetups-streaming-api-with-rxnetty.html
