关于响应式编程的理解与SpringCloudGateway的理解
- 一. 响应式编程与函数式编程的区别
- 二. 响应式编程中常用的组件
- 2.1 RxJava定义
- 2.2 Rxjava基本概念
- 2.3 RxJava 用法
- 三 SpringcloudGateway
- 四 常见的四种限流规则
一. 响应式编程与函数式编程的区别
总的来说，响应式编程主要体现在①异步、②观察者模式
以webflux和servlet为例子
一个是每个api请求代表着一个线程。
另一个是一个主线程内，可以使用多个子线程，达到异步的效果。
二. 响应式编程中常用的组件
2.1 RxJava定义
RxJava是一个可以在JVM上运行的，基于观察者模式 实现异步操作的java库。
2.2 Rxjava基本概念
// 创建观察者
Observer observer = new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "onSubscribe");
}
@Override
public void onNext(String o) {
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "onError data is :" + e.toString());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
};
// 创建被观察者
Observable observable = Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(@NonNull ObservableEmitter e) throws Exception {
e.onNext("hello");
e.onNext("world");
e.onComplete();
}
});
// 订阅
observable.subscribe(observer);
2.3 RxJava 用法
- Rxjava 实现线程切换
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
//1、“异步线程” 执行耗时操作
//2、“执行完毕” 调用onNext触发回调，通知观察者
e.onNext("1");
e.onComplete();
}
}).subscribeOn(Schedulers.io()) // io线程
.observeOn(AndroidSchedulers.mainThread()) // 主线程
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
// 订阅线程 订阅的那一刻在订阅线程中执行
}
@Override
public void onNext(String value) {
// “主线程”执行的方法
}
@Override
public void onError(Throwable e) {
// "主线程"执行的方法
}
@Override
public void onComplete() {
// "主线程"执行的方法
}
});
- Rxjava 使用操作符
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
// IO 线程
// 请求网络数据
e.onNext("123456");
}
}).map(new Function<String, Integer>() {
@Override
public Integer apply(String s) {
// IO 线程
// 网络数据解析(数据转化)
//
// throw new RequestFailException("获取网络请求失败");
return 123;
}
}).doOnNext(new Consumer<Integer>() { //保存登录结果UserInfo
@Override
public void accept(@NonNull Integer bean) throws Exception {
// IO 线程
// 保存网络数据
}
}).subscribeOn(Schedulers.io()) //IO线程
.observeOn(AndroidSchedulers.mainThread()) //主线程
.subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer bean) throws Exception {
// 更新UI
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
// 错误 显示错误页面
}
});
- Flowable
Flowable是为了应对
Backpressure(背压)产生的。
Backpressure(背压) 即生产者的生产速度大于消费者的消费能力引起的问题。
异步线程中 生产者有无限的生产能力；
主线程中 消费者消费能力不足,从而造成事件无限堆积，最后导致OOM。
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
//1、“异步线程” 执行耗时操作
//2、“执行完毕” 调用onNext触发回调，通知观察者
emitter.onNext(0);
emitter.onComplete();
}
// 若消费者消费能力不足，则抛出MissingBackpressureException异常
}, BackpressureStrategy.ERROR)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
// 订阅时执行，发生在“订阅线程”
// 这个方法是用来向生产者申请可以消费的事件数量
// 这里表明消费者拥有Long.MAX_VALUE的消费能力
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
// “主线程”执行的方法
}
@Override
public void onError(Throwable t) {
// "主线程"执行的方法
}
@Override
public void onComplete() {
// "主线程"执行的方法
}
});
各种背压策略在此不做多说明。
三 SpringcloudGateway
是使用webFlux编写的网关组件。
webflux与rxjava的异步编程是两种哲学编程概念。但是都是体现在了异步与观察者模式。
关于SpringcloudGateway，我理解我们需要做的东西
-
导入springcloudgateway依赖与nacos依赖。
-
书写配置类
2.1 GatewayConfiguration，这个配置类用于定义全局异常与全局过滤器。
2.2 RateLimiterConfiguration，这个配置类用于定义限流规则。 -
具体有关限流的配置，在
nacos配置中心配置。
四 常见的四种限流规则
给大家推荐一篇文章
