高阶程序员必备技能：Fizz网关的二次开发

原创

openapplus

修改于 2020-10-30 14:14:53

1.1K0

修改于 2020-10-30 14:14:53

文章被收录于专栏：OpenApplusOpenApplus

一、概述

在使用 fizz 过程中，可能会碰到：

需要定制http serve
需要额外的http client
需要自定义http filte
需要访问mysql、redis/codis、mongo、kafka 等

等问题，下面依次介绍解决办法，同时其它二次开发问题亦可参考。

二、定制http serve

fizz 采用 webflux 官方默认亦是最优的 http server 实现，并通过 WebFluxConfig 暴露，以方便外界进行细粒度的控制。

不建议创建多个 http server，即使它们共享同一端口。

webflux 默认基于 reactor-netty 实现 http server，可通过 NettyReactiveWebServerFactory 进行定制和扩展，包括 tcp、http、reactor-netty、netty、系统资源等层面，fizz 的 WebFluxConfig 含 NettyReactiveWebServerFactory bean，可修改或创建新的 NettyReactiveWebServerFactory bean 以定制 http server，下面是创建NettyReactiveWebServerFactory bean 的例子。

    @Bean
    public NettyReactiveWebServerFactory nettyReactiveWebServerFactory(ServerProperties serverProperties) {
        NettyReactiveWebServerFactory httpServerFactory = new NettyReactiveWebServerFactory();
        httpServerFactory.setResourceFactory(null);
        LoopResources lr = LoopResources.create("fizz-el", 1, Runtime.getRuntime().availableProcessors(), true);
        httpServerFactory.addServerCustomizers(
                httpServer -> (
                        httpServer.tcpConfiguration(
                                tcpServer -> {
                                    return (
                                            tcpServe
                                                    .runOn(lr, false) // 指定运行 server 的 eventloop 资源
                                                    .port(8080)       // server 监听的端口
                                                    .bootstrap(
                                                            serverBootstrap -> (
                                                                    // 下面定制 netty 并调整 tcp 参数
                                                                    serverBootstrap
                                                                            .option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT)
                                                                            .childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT)
                                                                            .childOption(ChannelOption.SO_KEEPALIVE, true)
                                                                            .childOption(ChannelOption.TCP_NODELAY,  true)
                                                            )
                                                    )
                                    );
                                }
                        )
                )
        );
        return httpServerFactory;
    }

netty 的默认配置适用绝大多数场景，尽量少调整。

三、创建额外的 http client

对外 http 交互，可直接使用 fizz 的 FizzWebClient 或 proxyWebClient，proxyWebClient 就是一个 org.springframework.web.reactive.function.client.WebClient，

FizzWebClient 也是基于 proxyWebClient，提供了与 eureka 注册中心服务交互的便利。

尽量共享 FizzWebClient 或 proxyWebClient 进行 http 操作，不建议引入 apache httpclient、feign 等 http 客户端，即使它们是异步、响应式的，确实需要创建额外的 WebClient 时，可参考 proxyWebClientConfig 的做法，然后尽量共享新建的 WebClient，例如：

    private ConnectionProvider getConnectionProvider() {
        return ConnectionProvider.builder("fizz-cp").maxConnections(2_000)
                                                    .pendingAcquireTimeout(Duration.ofMillis(6_000))
                                                    .maxIdleTime(Duration.ofMillis(40_000))
                                                    .build();
    }
    private LoopResources getLoopResources() {
        LoopResources lr = LoopResources.create("fizz-wc-el", Runtime.getRuntime().availableProcessors(), true);
        lr.onServer(false);
        return lr;
    }
    public WebClient webClient() {
        ConnectionProvider cp = getConnectionProvider(); // 客户端连接池
        LoopResources lr = getLoopResources();           // 运行客户端的 eventloop 资源
        HttpClient httpClient = HttpClient.create(cp).compress(false).tcpConfiguration(
                tcpClient -> {
                    return tcpClient.runOn(lr, false)
                                    // 定制客户端底层 netty
                                    // .bootstrap(
                                    //         bootstrap -> (
                                    //                 bootstrap.channel(NioSocketChannel.class)
                                    //         )
                                    // )
                                    .doOnConnected(
                                            connection -> {
                                                connection.addHandlerLast(new ReadTimeoutHandler( 20_000, TimeUnit.MILLISECONDS))
                                                          .addHandlerLast(new WriteTimeoutHandler(20_000, TimeUnit.MILLISECONDS));
                                            }
                                    )
                                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 20_000)
                                    .option(ChannelOption.TCP_NODELAY,            true)
                                    .option(ChannelOption.SO_KEEPALIVE,           true)
                                    .option(ChannelOption.ALLOCATOR,              UnpooledByteBufAllocator.DEFAULT);
                }
        );
        return WebClient.builder().exchangeStrategies(ExchangeStrategies.builder().codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(-1)).build())
                                  .clientConnector(new ReactorClientHttpConnector(httpClient)).build();
    }

WebClient 的配置和 http server 是类似的。

四、自定义 http filte

如果需要在请求处理的流水线上加入逻辑，可通过插件机制实现，具体可参考插件章节，尽量避免自定义 WebFilter，如果需要，应该继承 ProxyAggrFilter：

public abstract class ProxyAggrFilter implements WebFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        String serviceId = WebUtils.getServiceId(exchange); // 即请求的 path 是以 /proxy 开头
        if (serviceId == null) {
            return chain.filter(exchange);
        } else {
            return doFilter(exchange, chain);
        }
    }

    public abstract Mono<Void> doFilter(ServerWebExchange exchange, WebFilterChain chain);
}

实现 doFilter 方法即可，注意 filter 的执行顺序，需在 fizz 的 PreFilter 和 RouteFilter 之间。

五、访问 mysql、redis/codis、mongo、kafka 等

不建议在 fizz 中直接与 mysql 等传统数据库交互，因为它们没有原生的异步客户端，尽量把数据转移到分布式或本地缓存中，如 redis。

对 redis/codis、mongo、kafka 等操作，应使用 spring 官方提供的响应式客户端，注意客户端版本要与 spring boot 版本一致，

客户端使用可参官方文档，至于与 fizz 的整合，包括涉及多服务端的场景等，可参考 fizz 内置的 redis 交互逻辑设计和实现，RedisReactiveConfig 及子类AggregateRedisConfig。

比如有个 biz0 redis 库，在 fizz 中可按如下方式定义与其交互的逻辑：

在 application.yml 中加入：

biz0.redis.host: biz0 的 ip
biz0.redis.port: 6379
biz0.redis.password: 123456
biz0.redis.database: 0

定义对应 yml 的配置 bean，及与其交互的 redistemplate：

@Configuration
public class Biz0RedisConfig extends RedisReactiveConfig {
    
    static final String BIZ0_REACTIVE_REDIS_PROPERTIES         = "biz0ReactiveRedisProperties";
    static final String BIZ0_REACTIVE_REDIS_CONNECTION_FACTORY = "biz0ReactiveRedisConnectionFactory";
    static final String BIZ0_REACTIVE_REDIS_TEMPLATE           = "biz0ReactiveRedisTemplate";

    @ConfigurationProperties(prefix = "biz0.redis")
    @Configuration(BIZ0_REACTIVE_REDIS_PROPERTIES) // 此 bean 对应上面 yml 配置
    public static class biz0RedisReactiveProperties extends RedisReactiveProperties {
    }

    public Biz0RedisConfig(@Qualifier(BIZ0_REACTIVE_REDIS_PROPERTIES) RedisReactiveProperties properties) {
        super(properties);
    }

    @Override
    @Bean(BIZ0_REACTIVE_REDIS_CONNECTION_FACTORY)
    public ReactiveRedisConnectionFactory lettuceConnectionFactory() {
        return super.lettuceConnectionFactory();
    }

    @Override
    @Bean(BIZ0_REACTIVE_REDIS_TEMPLATE) // 与 biz0 redis 交互的 template
    public ReactiveStringRedisTemplate reactiveStringRedisTemplate(
            @Qualifier(BIZ0_REACTIVE_REDIS_CONNECTION_FACTORY) ReactiveRedisConnectionFactory factory) {
        return super.reactiveStringRedisTemplate(factory);
    }
}

RedisReactiveConfig 是 fizz 抽象的通用 redis/codis 配置，支持多服务端场景，并能在各对应客户端间共享资源：

public abstract class RedisReactiveConfig {

    protected static final Logger log = LoggerFactory.getLogger(RedisReactiveConfig.class);

    // this should not be changed unless there is a truly good reason to do so
    private static final int ps = Runtime.getRuntime().availableProcessors();
    private static final ClientResources clientResources = DefaultClientResources.builder()
            .ioThreadPoolSize(ps)
            .computationThreadPoolSize(ps)
            .build();

    private RedisReactiveProperties redisReactiveProperties;

    // 子类覆盖并定义配置
    public RedisReactiveConfig(RedisReactiveProperties properties) {
        redisReactiveProperties = properties;
    }

    // 子类覆盖，创建与特定 redis/codis 交互的 template
    public ReactiveStringRedisTemplate reactiveStringRedisTemplate(ReactiveRedisConnectionFactory fact) {
        return new ReactiveStringRedisTemplate(fact);
    }

    // 子类覆盖，指定客户端连接池，通常子类不用调整此逻辑
    public ReactiveRedisConnectionFactory lettuceConnectionFactory() {

        log.info("connect to " + redisReactiveProperties);

        RedisStandaloneConfiguration rcs = new RedisStandaloneConfiguration(redisReactiveProperties.getHost(), redisReactiveProperties.getPort());
        String password = redisReactiveProperties.getPassword();
        if (password != null) {
            rcs.setPassword(password);
        }
        rcs.setDatabase(redisReactiveProperties.getDatabase());

        LettucePoolingClientConfiguration ccs = LettucePoolingClientConfiguration.builder()
                .clientResources(clientResources)
                .clientOptions(ClientOptions.builder().publishOnScheduler(true).build())
                .poolConfig(new GenericObjectPoolConfig())
                .build();

        return new LettuceConnectionFactory(rcs, ccs);
    }
}