基于Rx-netty和Karyon2的云就绪微服务

Netflix Karyon提供了一个干净的框架来创建可用于云的微服务。在您的组织中，如果您使用包含Eureka的Netflix OSS堆栈进行服务注册和发现，使用Archaius进行资产管理，那么很可能会使用Karyon创建微服务。

Karyon最近发生了很多变化，我的目的是使用新版本的Karyon记录一个好的样本。旧的细胞核（称之为Karyon1）是基于JAX-RS 1.0规格与新泽西的实施，细胞核（Karyon2）的新版本仍然支持球衣也鼓励使用RX-的Netty这是一个定制版本的Netty用支持Rx-java 。

话虽如此，让我跳入一个样本。我对这个样本的目标是创建一个“乒乓”微服务，该服务接受一个“ POST”消息并返回一个“确认”消息。

以下是一个示例请求：

{
"id": "id",
"payload":"Ping"
}

和预期的响应：

{"id":"id","received":"Ping","payload":"Pong"}

第一步是创建一个RequestHandler ，顾名思义，它是一个用于处理传入请求路由的RX-Netty组件：

package org.bk.samplepong.app;import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.netty.protocol.http.server.HttpServerRequest;
import io.reactivex.netty.protocol.http.server.HttpServerResponse;
import io.reactivex.netty.protocol.http.server.RequestHandler;
import netflix.karyon.transport.http.health.HealthCheckEndpoint;
import org.bk.samplepong.domain.Message;
import org.bk.samplepong.domain.MessageAcknowledgement;
import rx.Observable;import java.io.IOException;
import java.nio.charset.Charset;public class RxNettyHandler implements RequestHandler<ByteBuf, ByteBuf> {private final String healthCheckUri;private final HealthCheckEndpoint healthCheckEndpoint;private final ObjectMapper objectMapper = new ObjectMapper();public RxNettyHandler(String healthCheckUri, HealthCheckEndpoint healthCheckEndpoint) {this.healthCheckUri = healthCheckUri;this.healthCheckEndpoint = healthCheckEndpoint;}@Overridepublic Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {if (request.getUri().startsWith(healthCheckUri)) {return healthCheckEndpoint.handle(request, response);} else if (request.getUri().startsWith("/message") && request.getHttpMethod().equals(HttpMethod.POST)) {return request.getContent().map(byteBuf -> byteBuf.toString(Charset.forName("UTF-8"))).map(s -> {try {Message m = objectMapper.readValue(s, Message.class);return m;} catch (IOException e) {throw new RuntimeException(e);}}).map(m -> new MessageAcknowledgement(m.getId(), m.getPayload(), "Pong")).flatMap(ack -> {try {return response.writeStringAndFlush(objectMapper.writeValueAsString(ack));} catch (Exception e) {response.setStatus(HttpResponseStatus.BAD_REQUEST);return response.close();}});} else {response.setStatus(HttpResponseStatus.NOT_FOUND);return response.close();}}
}

该流程是完全异步的，并由RX-java库在内部进行管理，Java 8 Lambda表达式也有助于使代码简洁。您将在这里看到的一个问题是路由逻辑（控制器的哪个uri）与实际的控制器逻辑混合在一起，我认为这已得到解决。

有了这个RequestHandler，就可以使用原始的RX-Netty在独立的Java程序中启动服务器，实质上就是这样，将在端口8080处启动一个端点来处理请求：

public final class RxNettyExample {public static void main(String... args) throws Exception {final ObjectMapper objectMapper = new ObjectMapper();RxNettyHandler handler = new RxNettyHandler();HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(8080, handler);server.start();

但是，这是本机的Rx-netty方法，对于支持云的微服务，必须进行一些操作，该服务应在Eureka上注册，并应响应Eureka的运行状况检查，并应该能够使用Archaius加载属性。

因此，对于Karyon2，主程序中的启动看起来有点不同：

package org.bk.samplepong.app;import netflix.adminresources.resources.KaryonWebAdminModule;
import netflix.karyon.Karyon;
import netflix.karyon.KaryonBootstrapModule;
import netflix.karyon.ShutdownModule;
import netflix.karyon.archaius.ArchaiusBootstrapModule;
import netflix.karyon.eureka.KaryonEurekaModule;
import netflix.karyon.servo.KaryonServoModule;
import netflix.karyon.transport.http.health.HealthCheckEndpoint;
import org.bk.samplepong.resource.HealthCheck;public class SamplePongApp {public static void main(String[] args) {HealthCheck healthCheckHandler = new HealthCheck();Karyon.forRequestHandler(8888,new RxNettyHandler("/healthcheck",new HealthCheckEndpoint(healthCheckHandler)),new KaryonBootstrapModule(healthCheckHandler),new ArchaiusBootstrapModule("sample-pong"),KaryonEurekaModule.asBootstrapModule(),Karyon.toBootstrapModule(KaryonWebAdminModule.class),ShutdownModule.asBootstrapModule(),KaryonServoModule.asBootstrapModule()).startAndWaitTillShutdown();}
}

现在它基本上已经可以使用云了，该版本的程序在启动时将在Eureka进行干净注册并公开运行状况检查端点。它还在端口8077处公开了一套简洁的管理端点。

结论

我希望这对使用Karyon2开发基于Netflix OSS的内容有很好的介绍。整个示例可在我的github仓库中找到：https：//github.com/bijukunjummen/sample-ping-pong-netflixoss/tree/master/sample-pong。作为后续，我将展示如何使用spring-cloud开发相同的服务，这是Spring创建微服务的方式。