RxJava+SpringBoot使用场景：统计语数英及格的人数

需求描述

统计语数英及格的人数

• 数据库设计
  select * from grade;

代码处理

controller

• 传统curd方案

   @GetMapping("/passedCount")
    public Map<String, Long> passedCount(){
        return gradeRxService.studnetPassedCount();
    }

• RxJava方案

    // controller
    @GetMapping(value = "/passedCountObservable")
    public Flowable passedCountObservable(){
        return gradeRxService.passedCountObservable();
    }
// 返回的Flowable还需要解析成对应的对象才能返回给前端
@Configuration
@ComponentScan(basePackages = { "com.muyf.alibaba.curd.*" })
public class WebConfig extends WebMvcConfigurerAdapter {
    @Override
    public void addReturnValueHandlers(List<HandlerMethodReturnValueHandler> returnValueHandlers) {
        returnValueHandlers.add(new ObservableReturnValueHandler());
        returnValueHandlers.add(new FlowableReturnValueHandler());
    }
}
public class FlowableReturnValueHandler implements AsyncHandlerMethodReturnValueHandler {
    @Override
    public boolean isAsyncReturnValue(Object returnValue, MethodParameter returnType) {
        return returnValue != null && supportsReturnType(returnType);
    }

    @Override
    public boolean supportsReturnType(MethodParameter returnType) {
        return Observable.class.isAssignableFrom(returnType.getParameterType());
    }

    @Override
    public void handleReturnValue(Object returnValue, MethodParameter returnType, ModelAndViewContainer mavContainer,
                                  NativeWebRequest webRequest) throws Exception {

        if (returnValue == null || !Flowable.class.isInstance(returnValue))  {
            mavContainer.setRequestHandled(true);
            return;
        }

        final Flowable<?> flowable = Flowable.class.cast(returnValue);
        WebAsyncUtils.getAsyncManager(webRequest)
                .startDeferredResultProcessing(new FlowableAdapter<>(flowable), mavContainer);
    }

    private static class FlowableAdapter<T> extends DeferredResult<T> {
        public FlowableAdapter(Flowable<T> flowable) {
            flowable.subscribe(this::setResult, this::setErrorResult);
        }
    }
}

service层

-- 传统curd方案

   @Override
    public Map<String, Long> studnetPassedCount() {
        Map<String,Long> result = new ConcurrentHashMap<>();
        for (String courseName : CourseEnum.courseArray) {
            List<Grade> grades = gradeMapper.selectStudentCountGreatThanGradeNumInCourse(courseName, Long.valueOf(60));
            result.put(courseName,Long.valueOf(grades.size()));
        }
        return result;
    }

• RxJava方案

    @Override
    public Flowable<GradePassDTO> passedCountObservable() {
        return Flowable.fromArray(CourseEnum.courseArray.toArray()).parallel().runOn(Schedulers.io()).map(course -> {
            System.out.println(course.toString()+Thread.currentThread());
            /* 打印结果
            *英语Thread[RxCachedThreadScheduler-3,5,main]
            *语文Thread[RxCachedThreadScheduler-2,5,main]
            *数学Thread[RxCachedThreadScheduler-4,5,main]
            */
            List<Grade> grades = gradeMapper.selectStudentCountGreatThanGradeNumInCourse(course.toString(), Long.valueOf(60));
            GradePassDTO dto = new GradePassDTO(course.toString(),Long.valueOf(grades.size()));
            return dto;
        }).sequential();
    }

Mapper层

<select id="selectStudentCountGreatThanGradeNumInCourse" resultType="com.muyf.alibaba.curd.domain.grade.Grade">
        select <include refid="Base_Column_List"/> from grade where grade_course = #{courseName} and grade_num > #{gradeNum};
</select>

实现效果

• 传统curd方案

  
  image.png

• RxJava方案

  
  image.png

代码分析

• 数据持久层处理方式不变
• 逻辑service层如果通过多次查询合并返回给前端的情况下适用Flowable来进行并发提高响应速度。
• controller层对返回的Flowable/abservable要做层解析然后返回给前端