vlambda博客
学习文章列表

响应式编程（reactive programming）

响应式编程（Reactive programming）是一种面向数据流和变化传播的编程范式。通俗理解就是在与外部资源交互时从传统的命令式编码风格转换为异步、非阻塞、函数式的编码风格；

• Responsive: 可响应的。要求系统尽可能做到在任何时候都能及时响应。

• Resilient: 可恢复的。要求系统即使出错了，也能保持可响应性。

• Elastic: 可伸缩的。要求系统在各种负载下都能保持可响应性。

• Message Driven:消息驱动的。要求系统通过异步消息连接各个组件。

电子表格程序就是响应式编程的一个例子。单元格可以包含字面值或类似"=A1+A2"的公式，而包含公式的单元格的值会依据其他单元格的值的变化而变化

常见的使用场景：

• Live database queries

• Mobile (Android)

• Big Data

• Real time analytics

• HTTP/2

异步调用常用的处理方式：

• Callback 多层嵌套之后，容易写出 回调地狱，严重影响可读性

  

• Future 用Promise模式解决回调问题

  
  

  

• ReactiveX 用响应式编程方式解决回调问题

  

• Coroutine 用类似Go语言的Goroutine方式解决回调问题，因为多了一步暂停（suspend）和恢复（resume）所以性能会比回调方式略低。 

Reactive Streams

• Reactive Streams 是为支持非阻塞背压(backpressure)的异步流处理提供标准.

• Co-designed by Netflix, Lightbend, Pivotal, RedHat, Kaazing, Twitter ...

• 主流实现 RxJava2, RxJs, Reactor3, Java9Flow ...

• 四个标准接口+一个TCK测试验证工具包

几种Jvm实现的对比：

Reactor Core 3

目前的主流reactive框架

线程模型对比

传统的servlet： 

reactive： 

spring5 新特性：

Flux Mono

Spring Webflux

spring mvc实现：

reactive web：

• ServerHttpRequest ServerHttpResponse

  
  

  不同的服务引擎都有对应的reactive adapter

  

• HttpHandler

  
  

  传递相应的request response 到ServerWebExchage 

  

• WebHandler WebFilterChain WebFilter

  
  

  走完所有的filter 调用WebHandler 

  

• HandlerMapping HandlerAdapter

  
  

RouterFunction vs Annotation

可以支持多种写法：java类似

Webclient 一种可替代RestTemplate跨服务通信方式

可以组合各种不同的API，在微服务中可以使用 

来个crud demo： 

Reactive data access with Spring Data

Spring Data 2.1 目前支持这几个NoSQL数据库 Reactive connectors, MongoDB, Cassandra, Couchbase, and Redis；

PostgreSQL 已经有几个可选的Reactive connectors；

MySQL and MariaDB 暂时还不支持；

nosql数据库貌似更主流的支持Reactive；

AsynchronousDatabaseAccess (ADBA) java.sql2的一个实现，目前还是实验状态，

ReactiveRelationalDatabaseConnectivity (R2DBC) Spring Data官方维护的一个新的框架；包括 R2DBC SPI, R2DBCClient， R2DBCPostgreSQL，不知道未来会不会支持mysql版本；spring data r2dbc: 

jdbc vs rjdbc 

同步CrudRepository包装成reactive模式

reactor中的同步调用必须使用Scheduler

## 性能测试对比

测试工具： Gatling

mvc-with-latency vs webflux-with-latency：

 
   
   
 

  
    
    
  
mvn gatling:test
  
    
    
  
-Dgatling.simulationClass=LoadSimulation 
  
    
    
  
-Dbase.url=http://localhost:8091/
  
    
    
  
-Dtest.path=hello/100 
  
    
    
  
-Dsim.users=1000 
  
    
    
  
-Dtest.name=mvc-with-latency
  
    
    
  


  
    
    
  
mvn gatling:test
  
    
    
  
-Dgatling.simulationClass=LoadSimulation
  
    
    
  
-Dbase.url=http://localhost:8092/
  
    
    
  
-Dtest.path=hello/100 -Dsim.users=300 
  
    
    
  
-Dtest.name=webflux-with-latency
 
   
   
 

测试数据如下：

• 线程数是指 jconsole里面的峰值线程数

• 这里没有统计 webflux线程数量，因为对于运行在异步IO的 Netty之上的 WebFlux应用来说，其工作线程数量始终维持在一个固定的数量上，默认固定的数量等于 CPU核数（*2）；

• 随着用户数的增多，吞吐量基本呈线性增多的趋势；

• 95%的响应都在100ms+的可控范围内返回了，并未出现延时的情况。

mvc-with-latency：

由以上数据可知：

• 用户量在接近3000的时候，线程数达到默认的最大值200；

• 线程数达到200前，95%的请求响应时长是正常的（比100ms多一点点），之后呈直线上升的态势；

• 线程数达到200后，吞吐量增幅逐渐放缓。

• 当所有可用线程都在阻塞状态的话，后续再进入的请求只能排队，从而当达到最大线程数之后，响应时长开始上升.

• 由于工作线程数扩大一倍，因此请求排队的情况缓解一半。

增加线程数确实可以一定程度下提高吞吐量，降低因阻塞造成的响应延时，但是增加线程会带来更多的内存消耗以及线程上下文切换成本；

对比数据：

最后测试了一下20000用户的情况：

• mvc-with-latency 自然是抗不住的，fail；

• WebFlux-with-latency 吞吐量达到7000+ req/sec，95%响应时长仅100ms左右；