案例分析 -- 反应式编程框架 Flower 的设计

程序运行： 普通场景

1. 请求到达：用户请求服务器，服务器使用 Web 端口（比如：80）服务用户请求。

2. 1 请求/1 线程：Web 容器（比如 Tomcat）为每个用户请求启动一个线程，由这个线程同步处理全部请求。

3. 数据库访问：再数据库连接池中获取数据库连接，访问数据库。

4. 访问数据库物理文件（磁盘 IO）

程序同步阻塞：

 -1. 数据未到达而被阻塞：用户请求使用网络连接到服务器 Web 端口，由于网络原因，后者客户端原因，数据发送比较慢，

       还没有到达服务器 Web 端口，线程从网络连接中读取数据，读取不到数据，就会等待数据而被阻塞。

0. 线程同步访问被阻塞：通过网络同步访问第三方服务，同步访问数据库而被阻塞。  

1. 等待数据库连接而被阻塞：数据库连接数量可能小于线程数量，多个线程同时争抢数据库连接，没有抢到数据库连接的线程就会被阻塞。

2. 数据库请求排队阻塞：数据库连接发送请求通过网络连接到达数据库，多个请求需要排队

3. 访问文件系统（磁盘 IO 操作）而被阻塞： 访问数据库的文件系统,因 IO 操作而被阻塞 

 程序崩溃：高并发，高负载场景

1. 高并发请求同时到达：大量用户请求同时到达服务器。

2. Web 容器创建大量线程：线程数量急剧上升。创建线程是个昂贵操作，需要耗费服务器资源（CPU，内存），到达一定数量，耗尽服务器资源，无法再创建更多的线程

3. 后续请求等待： 后续到达的用户请求需要等待，超时后，仍然无法处理，从用户请求的角度看，服务器不可用。

4. Web 容器内线程争夺资源被阻塞：Web 容器内大量线程为了完成服务，争抢资源，资源不足时，线程因为等待其他资源而被阻塞。线程不释放自己的资源，导致后来的线程无资源可用也会被阻塞====>整个系统响应变慢。

5. 硬件资源不足导致系统越来越慢：CPU，内存，带宽不足时，系统就会越来越慢。

6. 大量的用户请求无法得到处理，系统就崩溃啦。

分析崩溃原因:

     核心原因：阻塞。 阻塞导致线程不能释放自己的资源，其他线程无资源可用，自己因被阻塞而无法向下运行。后来的请求，无法被处理，最终导致系统崩溃。简言之：阻塞导致线程争抢资源，系统资源被耗尽。

解决方法：解除阻塞----异步。

解决方案：反应式编程框架----Flower。

异步关注点：如何改善系统性能？如何实现异步的？编程如何实现的？编程框架如何设计的？

解析：

1.请求并发到达服务器，访问 Web 端口（80 端口），每个请求建立 Socket 连接，这些连接等待 Web 容器处理请求。

2.Web 容器使用有限的几个线程（容器线程），甚至一个线程，处理所有的用户请求连接---多路复用

   ===>多路 socket 的请求使用一个或者几个有限的线程，遍历多个 socket 网络连接，为他们进行网络通信操作。

   有限的线程可以复用更多的并发请求。

3.容器线程从 socket 连接中抽取 request 和 response 对象，然后将 request 和 response 对象交给后面的应用程序处理。

   socket 的连接中如果没有数据，就没有容器线程处理这个 connection。===>没有线程阻塞。

   socket 的 connection 连接中如果有数据，容量线程将数据抽取为 request 和 response 对象，交给后面的应用程序。

   ==>有限的容器线程，就可以处理很多个并发连接===>避免无限制创建线程，导致系统资源耗尽。

4.Flower 运行环境有自己的线程。提供自己的线程池。容器线程提交 request 和 response 数据到 Flower 后，

   Flower 将数据交给 Service 处理。线程再某个 service 上执行。

5.ServiceA 处理完后，调用 ServiceB。

  传统方式--同步调用：ServiceA 同步调用 ServiceB，ServiceB 进行 IO 操作被阻塞，ServiceA 也会被阻塞，不能往下执行。ServiceA 和 ServiceB 都被阻塞。

  Flower 异步调用：ServiceA 处理完成后，发送消息给 ServiceB。ServiceB 使用 akka 线程执行自己，

                           拿到 ServiceA 传递过来的消息，进行处理。完成后，发送消息给 ServiceC。

  Flower 中并发执行：ServiceA，ServiceB，ServiceC 并发执行。ServiceA 处理完 A 用户 request 的第一步后，发送消息给 ServiceB，ServiceB 执行的时候；ServiceA 开始处理 B 用户的 request。

                               ===>ServiceA 和 ServiceB 使用不同的线程同时处理。

6.异步数据库访问：

         传统方式：获取数据库连接，getConnection()，如果没有可用数据库连接，就会被阻塞。

         Flower 方式：异步数据库连接驱动，返回的是 Future，知道 Future 完成操作 Complete，

                            由另外一个线程拿到 Connection 提交到数据库==>获取数据库连接的过程是不阻塞的。

效果解析：因为线程少，不会造成很多的资源消耗；因为不阻塞，提高系统的并发量

                ==>整体的性能很好，系统的吞吐量 TPS 和响应时间表现都很好。

Flower 实现异步的基础是 Akka 的 Actor：

原理解析：生产者和消费者模式，使用消息队列，分离生产者和消费者。===>解决直接调用的阻塞问题。

                Service 被封装在 Actor 中，Actor 之间相互通信。

1. Sender 发送消息给一个 Actor。

       同步调用：Sender 直接调用 Actor，Actor 自己被阻塞，Sender 也会被阻塞。

       异步调用：使用消息机制。Sender 不直接调用 Actor,而是发送消息 Message 给 Actor。

    2.消息发送：Actor 在自己门牌号前挂了一个邮箱（队列），发送过来的消息，先入队放入 MailBox 邮箱。

    3.消息分派：Dispatcher 分发器定时分派分发邮箱中的消息给 Actor

    4.Actor 处理消息

    5.处理完成后，封装消息发送到下一个 Actor。

Akka 示例：

    public class Hello extends UntypedActor{

          public void onReceive(Object message){

                 System.out.println("Hello,world");

          } 

   }

   ActorRef hello=...;

   hello.tell("hi!");

Service 开发示例：

      //1.开发异步编程 Service

       public class ServiceA implements Service<String,String> {

             public String process(String message,ServiceContext context){

                    if(message!=null && message intanceof String){

                         return ((String)message).trim();

                    }

                    return "";

             }

       }

      //2.编排 Service 流程，serviceA->serviceB->serviceC

        ServiceFlow.getOrCreate(flowName,serviceFactory)

                          .buildFlow(ServiceA.class,ServiceB.class)

                          .buildFlow(ServiceB.class,ServiceC.class);

     //3.调用异步服务：

        flowerFactory.getServiceFacade().asyncCallService(flowName,"hello world");

兼容 Spring 的 Flower Web 开发：

@RestController

@RequestMapping("/order/")

@Flower(value="createOrderFlow",flowNumber=32)

public class CreateOrderController extends FlowerController{

     @RequestMapping(value="createOrder") 

     public void createOrder(OrderDTO orderDTO,HttpServletRequest request)throws exception{

          doProcess(orderDTO);

     }

     protected void doProcess(Object param,HttpServletRequest request)throws IOException{

         AsyncContext context=null;

         if(request!=null){ context=req.startAsync();}

         flowRouter.asncCallService(param,context);

     } 

}

Flower 异步访问数据库：

@FlowerService

public class UserService implements Service<User,CompletableFuture<Serializable>>{

       @Autowired

       private UserDTO userDTO;

       @Override

       public CompletableFuture<Serializable> process(User message,ServiceContext context)throws throwable{

            System.out.println("处理用户信息："+message);

            return userDTO.insert(message);

       }

}

Flower 核心模块设计：