Статьи

Готовые к работе в облаке микросервисы с RX-Netty и Netflix Karyon

Ноябрь 2, 2017

Netflix Karyon предоставляет чистую платформу для создания облачных микросервисов. В вашей организации, если вы используете стек Netflix OSS, состоящий из Eureka для регистрации и обнаружения услуг , Archaius для управления имуществом, то весьма вероятно, что вы используете Karyon для создания своих микросервисов.

В последнее время Кэрион претерпел довольно много изменений, и моя цель здесь состоит в том, чтобы задокументировать хороший образец с использованием более новой версии Кэриона. Старый Karyon (назовите его Karyon1) был основан на спецификациях JAX-RS 1.0 с Джерси в качестве реализации, более новая версия Karyon (Karyon2) все еще поддерживает Джерси, но также поощряет использование RX-Netty, который является настроенной версией Netty с поддержка Rx-Java .

С учетом сказанного, позвольте мне перейти к образцу. Моя цель в этом примере — создать микро-сервис «pong», который принимает сообщение «POST» ed «и возвращает» Подтверждение «

Ниже приведен пример запроса:

{
"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;
    }

    @Override
    public 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 также помогают сделать код лаконичным. Единственная проблема, которую вы видите здесь, заключается в том, что логика маршрутизации (от того, к какому контроллеру) смешана с реальной логикой контроллера, и я считаю, что это решается .

Принимая во внимание этот RequestHandler, сервер может быть запущен в автономной Java-программе, используя сырую RX-Netty таким образом, это по сути, конечная точка будет установлена ​​на порту 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();

This is however the native Rx-netty way, for a cloud-ready micro-service a few things have to happen, the service should register with Eureka and should respond to the healthchecks back from Eureka and should be able to load up properties using Archaius.

So with Karyon2, the startup in a main program looks a little different:

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();
    }
}

Now it is essentially cloud ready, this version of the program on startup would register cleanly with Eureka and expose a healthcheck endpoint. It additionally exposes a neat set of admin endpoints at port 8077.

Conclusion

I hope this provides a good intro on using Karyon2 to develop Netflix OSS based. The entire sample is available at my github repo here: https://github.com/bijukunjummen/sample-ping-pong-netflixoss/tree/master/sample-pong. As a follow up I will show how the same service can be developed using spring-cloud which is the Spring way to create Micro-services.

36
Share: