响应式编程实战(02)-响应式编程的适用场景

news2024/11/16 7:53:52

0 知识前提

已掌握响应式编程中的核心概念：响应式流、背压机制以及响应式流规范。

1 引言

响应式编程能够应用到那些具体的场景呢?

目前有哪些框架中使用到了这一新型的技术体系呢?

2 响应式编程的应用场景分析

可以认为响应式编程并不仅仅是一种编程技术，更是一种架构设计的系统方法，因此可以应用于任何地方。

既可以用于简单的 Web 应用系统，也可以用于大型企业解决方案。

数据流处理是响应式编程的一大应用场景，流式系统的主要特点：

低延迟
高吞吐量

通过使用非阻塞式通信，可确保资源得到高效利用，实现低延迟和高吞吐量。

针对高并发流量，通常涉及大量 I/0 操作，相比传统的同步阻塞式 I/0 模型，响应式编程所具备的异步非阻塞式I/0 模型非常适合应对处理高并发流量的业务场景。网关的作用是用来响应来自前端系统的流量并将其转发到后端服务。

核心诉求

构建一个具有异步非阻塞式的请求处理流程的 Web 服务，需要高效处理跨服务之间的网络请求。

响应式编程在日常开发过程中日益得到广泛的应用：如Netflix Hystrix、Spring Cloud Gateway 及 Spring WebFlux，解析这些框架背后所应用的响应式编程技术。

3 响应式流规范

3.1 Netflix Hystrix 中的滑动窗口

Spring Cloud Netflix Hystrix 组件基于 Netflix Hystrix 实现了服务熔断功能

Netflix Hystrix，Netflix 开源的一款容错库，使用了HystrixCircuitBreaker 类实现熔断器。

HystrixCircuitBreaker 如何动态获取系统运行时的各项数据呢?

HealthCountsStream 设计上采用滑动窗口 (Rolling Window) 机制，大量采用数据流处理方面的技术及 RxJava 响应式编程框架。Hystrix 以秒为单位统计系统中所有请求的处理情况，然后每次取最近 10 秒的数据来进行计算，如果失败率超过一定闯值，就进行熔断。

如何实现这个滑动窗口？

把系统运行时所产生的所有数据都视为一个个的事件，这样滑动窗口中每个桶的数据都来自源源不断的事件，通常需要对其进行转换以便进行后续的操作。

Hystrix 采用基于响应式编程思想的 RxJava。使用 RxJava 的一大好处，可通过 RxJava 的一系列操作符来实现滑动窗口：

window 操作符，把当前流中的元素收集到另外的流序列
flatMap 操作符，把流中的每个元素转换成一个流，再把转换之后得到的所有流中的元素进行合并
reduce 操作符，对流中包含的所有元素进行累积操作，得到一个包含计算结果的流

Hystrix 巧妙运用 RxJava 的 window、flatMap等操作符来将单位窗口时间内的事件。以及将一个窗口大小内的 Bucket 聚集到一起形成滑动窗口，并基于滑动窗口集成指标数据。

3.2 Spring Cloud Gateway 中的过滤器

Spring 官方自己开发的一款 API 网关，基于最新的Spring5和Spring Boot2以及用于响应式编程的Proiect Reactor框架提供的是响应式、非阻塞式I/0 模型。

只需实现GlobalFilter接口，重写 filter()即可。

public class IPLimitFilter implements GlobalFilter

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        // 1. 获取当前的请求路径
        String url = exchange.getRequest().getURI().getPath();

        // 2. 获得所有的需要进行ip限流校验的url list
        List<String> ipLimitList = excludeUrlProperties.getIpLimitUrls();

        // 3. 校验并且判断
        if (ipLimitList != null && !ipLimitList.isEmpty()) {
            for (String limitUrl : ipLimitList) {
                if (antPathMatcher.matchStart(limitUrl, url)) {
                    // 如果匹配到，则表明需要进行ip的拦截校验
                    log.info("IPLimitFilter - 拦截到需要进行ip限流校验的方法：URL = " + url);
                    return doLimit(exchange, chain);
                }
            }
        }

        // 4. 默认直接放行
        return chain.filter(exchange);
    }
}

filter()返回了一个 Mono 对象，它就是在响应式编程框架 Project Reactor 中代表单个返回值的流式对象。

案例

@Component
public class PreGatewayFilterFactory extends AbstractGatewayFilterFactory<PreGatewayFilterFactory.Config> {

	public PreGatewayFilterFactory() {
		super(Config.class);
	}

	@Override
	public GatewayFilter apply(Config config) {
		// grab configuration from Config object
		return (exchange, chain) -> {
			//If you want to build a "pre" filter you need to manipulate the
			//request before calling chain.filter
			ServerHttpRequest.Builder builder = exchange.getRequest().mutate();
			//use builder to manipulate the request
			return chain.filter(exchange.mutate().request(builder.build()).build());
		};
	}

	public static class Config {
		//Put the configuration properties for your filter here
	}

}