首页 > > 程序设计 > Java技术 >

Spring WebFlux之HttpHandler的探索

2019-08-16 10:27:13来源：博客园 阅读 ()

新老客户大回馈,云服务器低至5折

Spring WebFlux之HttpHandler的探索

这是本人正在写的《Java 编程方法论：响应式Reactor3、Reactor-Netty和Spring WebFlux》一书的文章节选，它是《Java编程方法论：响应式RxJava与代码设计实战》的续篇，也可作为独立的一本来读

这是此节上半段的节选内容

HttpHandler的探索

通过前面的章节，我们已经接触了Reactor-Netty整个流程的设计实现细节，同时也涉及到了reactor.netty.http.server.HttpServer#handle，准确得说，它是一个SPI(Service Provider Interface)接口，对外提供BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler，这样，我们可以针对该handler依据自身环境进行相应实现。
Spring WebFlux与Reactor-Netty都有一套属于自己的实现，只不过前者为了适应Spring Web的一些习惯做了大量的适配设计，整个过程比较复杂，后者提供了一套简单而灵活的实现。那么本章我们就从Reactor-Netty内对它的实现开始，正式向Spring WebFlux进行过渡。

HttpServerRoutes设定

往往我们在给后台服务器提交HTTP请求的时候，往往会涉及到get，head，post，put这几种类型，还会包括请求地址，服务端会根据请求类型和请求地址提供对应的服务，然后才是具体的处理，那么我们是不是可以将寻找服务的这个过程抽取出来，形成服务路由查找。

于是，在Reactor-Netty中，设计了一个HttpServerRoutes接口，该接口继承了BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>>，用来路由请求，当请求来临时，对我们所设计的路由规则按顺序依次查找，直到第一个匹配，然后调用对应的处理handler。HttpServerRoutes接口内针对于我们常用的get，head，post，put，delete等请求设计了对应的路由规则（具体请看下面源码）。

我们在使用的时候首先会调用HttpServerRoutes#newRoutes得到一个DefaultHttpServerRoutes实例，然后加入我们设计的路由规则，关于路由规则的设计，其实就是将一条条规则通过一个集合管理起来，然后在需要时进行遍历匹配即可，这里它的核心组织方法就是reactor.netty.http.server.HttpServerRoutes#route，在规则设计完后，我们就可以设计对应每一条规则的BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>>函数式实现，最后，当请求路由匹配成功，就可以调用我们的BiFunction实现，对请求进行处理。

//reactor.netty.http.server.HttpServerRoutes
public interface HttpServerRoutes extends
                                  BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>> {

    static HttpServerRoutes newRoutes() {
        return new DefaultHttpServerRoutes();
    }


    default HttpServerRoutes delete(String path,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        return route(HttpPredicate.delete(path), handler);
    }

    ...

    default HttpServerRoutes get(String path,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        return route(HttpPredicate.get(path), handler);
    }

    default HttpServerRoutes head(String path,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        return route(HttpPredicate.head(path), handler);
    }

    default HttpServerRoutes index(final BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        return route(INDEX_PREDICATE, handler);
    }

    default HttpServerRoutes options(String path,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        return route(HttpPredicate.options(path), handler);
    }

    default HttpServerRoutes post(String path,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        return route(HttpPredicate.post(path), handler);
    }

    default HttpServerRoutes put(String path,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        return route(HttpPredicate.put(path), handler);
    }

    HttpServerRoutes route(Predicate<? super HttpServerRequest> condition,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler);

    ...

}

关于路由规则的设计，结合前面所讲，我们可以在HttpServerRoutes的实现类中设计一个List用来存储一条条的规则，接下来要做的就是将制定的规则一条条放入其中即可，因为这是一个添加过程，并不需要返回值，我们可以使用Consumer<? super HttpServerRoutes>来代表这个过程。对于请求的匹配，往往都是对请求的条件判断，那我们可以使用Predicate<? super HttpServerRequest>来代表这个判断逻辑，由于单条路由规则匹配对应的BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>>处理，那么我们是不是可以将这两者耦合到一起，于是reactor.netty.http.server.DefaultHttpServerRoutes.HttpRouteHandler就设计出来了：

//reactor.netty.http.server.DefaultHttpServerRoutes.HttpRouteHandler
static final class HttpRouteHandler
        implements BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>>,
                    Predicate<HttpServerRequest> {

    final Predicate<? super HttpServerRequest>          condition;
    final BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>>
                                                        handler;
    final Function<? super String, Map<String, String>> resolver;

    HttpRouteHandler(Predicate<? super HttpServerRequest> condition,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler,
            @Nullable Function<? super String, Map<String, String>> resolver) {
        this.condition = Objects.requireNonNull(condition, "condition");
        this.handler = Objects.requireNonNull(handler, "handler");
        this.resolver = resolver;
    }

    @Override
    public Publisher<Void> apply(HttpServerRequest request,
            HttpServerResponse response) {
        return handler.apply(request.paramsResolver(resolver), response);
    }

    @Override
    public boolean test(HttpServerRequest o) {
        return condition.test(o);
    }
}

这里可能需要对request中的参数进行解析，所以对外提供了一个可供我们自定义的参数解析器实现接口：Function<? super String, Map<String, String>>，剩下的condition与resolver就可以按照我们前面说的逻辑进行。

此时，HttpRouteHandler属于一个真正的请求校验者和请求业务处理者，我们现在要将它们的功能通过一系列逻辑串联形成一个处理流程，那么这里可以通过一个代理模式进行，我们在HttpServerRoutes的实现类中通过一个List集合管理了数量不等的HttpRouteHandler实例，对外，我们在使用reactor.netty.http.server.HttpServer#handle时只会看到一个BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>>实现，那么，所有的逻辑流程处理都应该在这个BiFunction的apply(...)实现中进行，于是，我们就有下面的reactor.netty.http.server.DefaultHttpServerRoutes实现：

//reactor.netty.http.server.DefaultHttpServerRoutes
final class DefaultHttpServerRoutes implements HttpServerRoutes {


    private final CopyOnWriteArrayList<HttpRouteHandler> handlers =
            new CopyOnWriteArrayList<>();
    ...
    @Override
    public HttpServerRoutes route(Predicate<? super HttpServerRequest> condition,
            BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        Objects.requireNonNull(condition, "condition");
        Objects.requireNonNull(handler, "handler");

        if (condition instanceof HttpPredicate) {
            handlers.add(new HttpRouteHandler(condition,
                    handler,
                    (HttpPredicate) condition));
        }
        else {
            handlers.add(new HttpRouteHandler(condition, handler, null));
        }
        return this;
    }

    @Override
    public Publisher<Void> apply(HttpServerRequest request, HttpServerResponse response) {
        final Iterator<HttpRouteHandler> iterator = handlers.iterator();
        HttpRouteHandler cursor;

        try {
            while (iterator.hasNext()) {
                cursor = iterator.next();
                if (cursor.test(request)) {
                    return cursor.apply(request, response);
                }
            }
        }
        catch (Throwable t) {
            Exceptions.throwIfJvmFatal(t);
            return Mono.error(t); //500
        }

        return response.sendNotFound();
    }
    ...
}

可以看到route(...)方法只是做了HttpRouteHandler实例的构建并交由handlers这个list进行管理，通过上面的apply实现将前面的内容在流程逻辑中进行组合。于是，我们就可以在reactor.netty.http.server.HttpServer中设计一个route方法，对外提供一个SPI接口，将我们所提到的整个过程定义在这个方法中（得到一个HttpServerRoutes实例，然后通过它的route方法构建规则，构建过程在前面提到的Consumer<? super HttpServerRoutes>中进行，最后将组合成功的HttpServerRoutes以BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>>的角色作为参数交由HttpServer#handle）。

另外，我们在这里要特别注意下，在上面DefaultHttpServerRoutes实现的apply方法中，可以看出，一旦请求匹配，处理完后就直接返回结果，不再继续遍历匹配，也就是说每次新来的请求，只调用所声明匹配规则顺序的第一个匹配。

//reactor.netty.http.server.HttpServer#route
public final HttpServer route(Consumer<? super HttpServerRoutes> routesBuilder) {
    Objects.requireNonNull(routesBuilder, "routeBuilder");
    HttpServerRoutes routes = HttpServerRoutes.newRoutes();
    routesBuilder.accept(routes);
    return handle(routes);
}

于是，我们就可以通过下面的Demo来应用上面的设计：

import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;

public class Application {

    public static void main(String[] args) {
        DisposableServer server =
                HttpServer.create()
                          .route(routes ->
                              routes.get("/hello",        <1>
                                         (request, response) -> response.sendString(Mono.just("Hello World!")))
                                    .post("/echo",        <2>
                                         (request, response) -> response.send(request.receive().retain()))
                                    .get("/path/{param}", <3>
                                         (request, response) -> response.sendString(Mono.just(request.param("param")))))
                          .bindNow();

        server.onDispose()
              .block();
    }
}

在<1>处，当我们发出一个GET请求去访问/hello时就会得到一个字符串Hello World!。

在<2>处，当我们发出一个 POST请求去访问 /echo时就会将请求体作为响应内容返回。

在<3>处，当我们发出一个 GET请求去访问 /path/{param}
时就会得到一个请求路径参数param的值。

关于SSE在这里的使用，我们可以看下面这个Demo，具体的代码细节就不详述了，看对应注释即可：

import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;
import reactor.netty.http.server.HttpServerRequest;
import reactor.netty.http.server.HttpServerResponse;

import java.io.ByteArrayOutputStream;
import java.nio.charset.Charset;
import java.time.Duration;
import java.util.function.BiFunction;

public class Application {

    public static void main(String[] args) {
        DisposableServer server =
                HttpServer.create()
                          .route(routes -> routes.get("/sse", serveSse()))
                          .bindNow();

        server.onDispose()
              .block();
    }

    /**
     * 准备 SSE response
     * 参考 reactor.netty.http.server.HttpServerResponse#sse可以知道它的"Content-Type" 
     * 是"text/event-stream"
     * flush策略为通过所提供的Publisher来每下发一个元素就flush一次
     */
    private static BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>> serveSse() {
        Flux<Long> flux = Flux.interval(Duration.ofSeconds(10));
        return (request, response) ->
            response.sse()
                    .send(flux.map(Application::toByteBuf), b -> true);
    }

    /**
     * 将发元素按照按照给定的格式由Object转换为ByteBuf。
     */
    private static ByteBuf toByteBuf(Object any) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        try {
            out.write("data: ".getBytes(Charset.defaultCharset()));
            MAPPER.writeValue(out, any);
            out.write("\n\n".getBytes(Charset.defaultCharset()));
        }
        catch (Exception e) {
            throw new RuntimeException(e);
        }
        return ByteBufAllocator.DEFAULT
                               .buffer()
                               .writeBytes(out.toByteArray());
    }

    private static final ObjectMapper MAPPER = new ObjectMapper();
}

原文链接:https://www.cnblogs.com/yinchen/p/11195415.html
如有疑问请与原作者联系

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有

上一篇：Spring学习笔记(一)

下一篇：springboot整合quarzt定时任务+xml格式

相关文章

IDC资讯： 主机资讯 注册资讯 托管资讯 vps资讯 网站建设

网站运营： 建站经验 策划盈利 搜索优化 网站推广 免费资源

网站联盟： 联盟新闻 联盟介绍 联盟点评 网赚技巧

行业资讯： 搜索引擎 网络游戏 电子商务 广告传媒

网络编程： Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它

服务器技术： Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护

软件技巧： 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷

网页制作： FrontPages Dreamweaver Javascript css photoshop fireworks Flash

程序设计： Java技术 C/C++ VB delphi

网络知识： 网络协议 网络安全 网络管理 组网方案 Cisco技术

操作系统： Win2000 WinXP Win2003 Mac OS Linux FreeBSD

热门词条
最新资讯
热门关注
热门标签