系列文章
Spring 响应式编程 随记 – C1 为什么选择响应式 Spring
Spring 响应式编程 随记 – C2 Spring 响应式编程基本概念 (一)
Spring 响应式编程 随记 – C2 Spring 响应式编程基本概念 (二)
Spring 响应式编程 随记 – C2 Spring 响应式编程基本概念 (三)
Spring 响应式编程 随记 – C2 Spring 响应式编程基本概念 (四)
Spring 响应式编程 随记 – C2 Spring 响应式编程基本概念 (五)
Spring 响应式编程 随记 – C3 响应式流 新的标准流 (一)
Spring 响应式编程 随记 – C3 响应式流 新的标准流 (二)
C1 为什么选择响应式 Spring
响应式编程 vs 命令式编程
如何具备即时响应性？
- 弹性：scalability 可伸缩的扩容减容
- 可伸缩的分布式系统有难度
- resilience 回弹性：故障可响应
消息驱动通信
针对IO实现更高资源利用率，应该使用异步非阻塞交互模型。
- 接受的消息到了之后做出响应，否则休眠。
- 组件以非阻塞方式发送消息
实现方法：消息代理服务器
价值：即时响应
形式：弹性和回弹性（elasticity and resilience）
手段：消息驱动
背压：工作负载管理，确保一个处理阶段不会压垮另一个
JVM 响应式框架：
- Akka
- Vert.x
- Spring Cloud
调用阻塞：阻塞式通信违背消息驱动原则。A调用B, B阻塞A。
以前旧有的一些不够完美的解决方法：
- 方法1：可以用 回调 （callback） 来实现跨组件通信。
interface ShoppingCardService{
void calculate(Input in, Consumer<Output> c);
}
class SyncServiceA implements ShoppingCardService{
public void calculate(Input value, Consumer<Output> c){
Output res = new Output(value);
c.accept(res); // no block action
}
}
class AsyncServiceB implements ShoppingCardService{
public void calculate(Input value, Consumer<Output> c){
new Thread(() -> {
// put this block action into a thread
Output res = otherService.getRes(value);
//...
c.accept(res);
}).start;
}
}
需要理解多线程，避免共享数据修改陷阱和回调地狱。
- 方法2:用
Future(java.util.concurrent.Future)
interface ShoppingCardService{
Future<Output> calculate(Input in);
}
class OrderService{
private final ShoppingCardService shoppingCardService;
void process(){
Input intput = getInput();
Future<Output> future = shoppingCardService.calculate(input);
//...
Output output = future.get();
//...
}
}
Future 类包装器检查是否有可用结果，能否以阻塞方式获取。对结果延迟获取，但最终还是要阻塞当前线程来和外部执行进行同步。
- 方法3: 更好的
CompletionStage和CompletableFuture。
interface ShoppingCardService{
CompletionStage<Output> calculate(Input in);
}
class OrderService{
private final ComponentB comb;
Intput in = getInput();
comb.calculate(in)
.thenApply(out1 -> { })
.thenCombine(out2 -> { })
.thenAccept(out3 -> { })
}
CompletionStage 类包装期提供了API进行链式调用，但 Spring4 MVC 不支持
- 方法4: Spring 4 里 的
ListenableFuture和AsyncRestTemplate
AsyncRestTemplate template = new AsyncRestTemplate();
// define the callback
SuccessCallback onSuccess = r -> {};
FailureCallback onFailure = e -> {};
String url = "http://xxxxx/api/xx";
ListenableFuture<?> res = template.getForEntity(url,SomeEntity.class);
res.addCallback(onSuccess,onFailure);
用一个单独的线程包装阻塞式网络调用来实现，依赖于 Servlet API，所以实现基于线程单次请求（thread-per-request)。Spring 5 的 WebClient 跨服务通信不阻塞，Servlet3.0之后也引入了新异步非阻塞功能。
JAVA的多线程共享CPU，内存也有较多消耗，上下文切换有成本，多开线程去解决阻塞不划算。
Netty响应式服务器解决上下文切换问题。
异步处理：
- 请求 --> 响应
- 无限元素数据流(响应式管道) 源<>处理<>转换<==>过滤