一、问题

站点收到并发请求需要串行处理，或者并行执行的业务某个环节需要串行处理。

二、分析

1，只能针对操作同一条数据的业务处理串行处理，整体必须并行处理。

2，因为执行速度很快，发生并发时，最好每一条都被执行，而不是返回“请稍后重试”，需要等待和重试。

3,支持分布式部署，同时有多个节点。

4,如果整体请求量特别大且处理需要时间，那么理应采用异步和队列。用队列消费的时候，应该也只是调取业务方法，队列本身不应该关心业务上面的事情。且队列消费应该也是并发的。

三、解决方案

方案一：数据库多版本并发控制（mvcc）

本质是利用数据库update串行执行机制。数据行加版本，假如有A、B两个事务针对同一条数据处理。

事务B修改版本为1的数据，发现修改了0行，知道同时有别人修改过数据，回滚。有的数据库有隐藏字段用于mvcc,不用自己维护一个版本字段。

缺点：

1，要求业务处理对数据有修改，否则如果业务处理只查询不需要修改数据就没法用了，不能为了修改而增加修改逻辑。不过如果只是查询而没有数据变更，一般也不需要串行执行。

2，这种方式依赖于事务，如果程序执行过程没有事务，最后一步发现版本变更了，,前面执行的逻辑无法回滚。

不推荐

方案二：依赖数据库锁（悲观锁）

当一个事务开始执行，读取数据的时候就加锁，让其他操作等待，实现整体业务操作是串行化的，一个业务操作开始~结束，另一个业务操作才开始。

通过锁解决方案一中要求业务处理支持事务的问题。锁需要设置超时时间，防止死锁

缺点：需要为行数据进行锁定。频繁的加锁、解锁，影响效率。

不推荐。

方案三：分布式锁

最好不在数据库上操作实现串行化，增大数据库压力，在程序执行过程中实现串行。比如可以通过redis分布式锁实现，具体请自行搜索，哈哈。不一定要用redis,任何方式实现了分布式锁都可以，哪怕自己弄个数据库来实现都行（redis也是数据库的一种）。

当然如果你是单机的，可以在程序中加线程锁控制，每种语言都有。但是不推荐，如果以后升级集群部署了，估计会忘记改。

前言

现实开发中，一个页面很少只有一个请求的，一般都有多个请求，有的需要串行，有的需要并行，使用传统的方法，如果有n个接口，我们就要设置n个接口回调，如果是串行的话，还需要在当前接口成功或失败的地方，调用下个一个请求，一个接着一个，真的是要逼死强迫症患者，而且代码可读性非常的差，新人来了往往要看上半天，不好维护且容易出错。

本文使用RxHttp请求框架作为案例演示，如果你不了解RxHttp，请查看30秒上手新一代Http请求神器RxHttp

请容许我再唠两句：RxHttp从4月中旬开始推广，在大家都对新的Http请求框架学不动或者懒得学的情况下，RxHttp依然收获了一大波粉丝，目前在Github上一经有415颗星，其中

RxHttp 一条链发送请求，新一代Http请求神器

Android 史上最优雅的实现文件上传、下载及进度的监听

这两篇文章更是得到了得到「玉刚说」及「刘望舒」微信公众号独家原创发布，我想，这也是对RxHttp的一种肯定，欢迎大家体验RxHttp，它优雅的写法及强大的功能，相信你一定会爱上它。

gradle依赖

   implementation 'com.rxjava.rxhttp:rxhttp:1.0.9'
   annotationProcessor 'com.rxjava.rxhttp:rxhttp-compiler:1.0.9' //注解处理器，生成RxHttp类
   implementation 'com.rxjava.rxlife:rxlife:1.0.7'  //页面销毁，关闭请求，非必须

   // if you use kotlin
   kapt 'com.rxjava.rxhttp:rxhttp-compiler:1.0.9'

接下来，我们正是开始。

并行

现在很多页面是这样的，上面是Banner条，Banner条下面是数据列表(假设是学生列表)，这样就涉及到两个接口，一个是获取Banner条数据，另一是获取学生列表数据，这两个接口没有任何关系，所以我们可以并行去实现

//Banner 的Observable对象                                                
Observable<Banner> bannerObservable = RxHttp.get("http://...")        
    .asObject(Banner.class);                                          
                                                                      
//学生的Observable对象                                                     
Observable<List<Student>> studentObservable = RxHttp.get("http://...")
    .asList(Student.class);            
                                                                      
//这里使用RxJava组合符中的merge操作符，将两个被观察者合并为一个                                
Observable.merge(bannerObservable, studentObservable)                 
    .as(RxLife.asOnMain(this)) //感知生命周期，自动关闭请求                        
    .subscribe(o -> {                                                 
        //请求成功，回调2次，一次是Banner数据，一次Student列表                   
        if (o instanceof Banner) {                                    
            //获取到banner数据                                             
        } else if (o instanceof List) {                               
            //获取到学生列表数据                                               
        }                                                             
    }, throwable -> {                                                 
        //出现异常                                                        
    }, () -> {                                                        
        //2个请求执行完毕，开始更新UI                                             
    });                                                               

可以看到，我们首先通过RxHttp类拿到Banner和Student的两个Observable对象，然后通过merge操作符，将两个Observable对象合并为一个，并订阅观察者，这样就能在onNext回调中拿到Banner和Student数据，并在onComplete回调中更新UI。

可是，这样就完了吗？熟悉RxJava的同学应该知道，RxJava在出现异常后并且回调到onError接口时，就会停止工作，那么如果Banner接口先出现异常，岂不是收不到Student信息了？是的，那么，我们应该如何去处理呢，其实很简单，RxJava为我们提供了异常捕获操作符，如：onErrorResumeNext和onErrorReturn，作用就是出现异常了，我们如何去补救它。在这，我们使用onErrorResumeNext操作符，代码如下

//Banner 的Observable对象                                                
Observable<Banner> bannerObservable = RxHttp.get("http://...")        
    .asObject(Banner.class)
    .onErrorResumeNext(Observable.empty()); //出现异常，发送一个空的Observable对象                                         
                                                                      
//学生的Observable对象                                                     
Observable<List<Student>> studentObservable = RxHttp.get("http://...")
    .asList(Student.class);            
                                                                      
//这里使用RxJava组合符中的merge操作符，将两个被观察者合并为一个                                
Observable.merge(bannerObservable, studentObservable)                 
    .as(RxLife.asOnMain(this)) //感知生命周期，自动关闭请求                        
    .subscribe(o -> {                                                 
        //请求成功，回调2次，一次是Banner数据，一次Student列表                   
        if (o instanceof Banner) {                                    
            //获取到banner数据                                             
        } else if (o instanceof List) {                               
            //获取到学生列表数据                                               
        }                                                             
    }, throwable -> {                                                 
        //出现异常                                                        
    }, () -> {                                                        
        //2个请求执行完毕，开始更新UI                                             
    });

上面我们只加了onErrorResumeNext(Observable.empty())这一行代码，Observable.empty()是一个不会发射任何事件的Observable对象。所以，这个时候如果Banner的Observable出现异常，就不会发射任何事件，Student 的Observable对象便可继续执行，只是在onNext回调中，就只能收到一次Student的回调(请求成功的话)，并且随后执行onComplete回调更新UI，这样就能保证即使Banner接口出错了，我们依然可以正常现实学生列表数据。说的抽象一点就是保证A接口不影响B接口，但是B可以影响A接口，如果要保证A、B两个接口互不影响，分别对A、B接口处理异常即可，如果有3个、4个甚至更多的请求，可以使用Observable.mergeArray操作符。

串行

接下来，看看我们串行，假设我们有这样一个需求，需要在注册完成后立即去登录，这种情况下，就只能串行去实现，在这，我们使用RxJava的flatMap这个操作符去实现

flatMap

RxHttp.postForm("http://...") //发送注册请求
    .add("userName", "zhangsan")
    .add("password", "123456")
    .asObject(Register.class)
    .flatMap((io.reactivex.functions.Function<Register, ObservableSource<User>>) register -> {
        //注册成功，拿到注册信息去登录，并返回User对象
        return RxHttp.get("http://...") //发送登录请求
                .add("userId", register.getUserId())
                .add("password", register.getPassword())
                .subscribeOnCurrent() //当前线程发送登录请求
                .asObject(User.class);
    })
    .as(RxLife.asOnMain(this)) //感知生命周期，自动关闭请求
    .subscribe(user -> {
        //注册并且登录成功，拿到用户信息
    }, throwable -> {
        //出现异常，注册失败或者登录失败
    });

注：RxHttp中的asXXX系列方法，内部会默认开启IO线程执行Http请求，所以我们在发送单个请求时，无需指定请求执行线程；然而在多个请求串行时，为提升效率，我们希望一个线程可以执行多个请求，故我们需要使用subscribeOnCurrent方法指定请求在当前线程执行。

可以看到，这里我们使用flatMap操作符，当注册成功，就会走到flatMap内部去登录，登录成功就会拿到User对象并回调观察者。

小结

看完你会发现，RxHttp做到了与RxJava的无缝连接，使用asXXX系列方法，就可以拿到一个Observable<T>对象，随后再结合RxJava的merge、flatMap，就可以优雅的实现的Http请求的串行及并行。如果你对RxJava有一定的了解，还可以实现很多有意思的功能，比如：为单个请求设置超时、请求失败自动重试n次等等。

最后，一切功劳都要归功于RxJava的强大，感谢RxJava，向它致敬！！！！