mybatisplus 流式查询

约定不等于承诺〃 2022-12-30 04:59 372阅读 0赞

Dao.class 方法

@Select("select ${ew.sqlSelect} from t_app_usermlq t ${ew.customSqlSegment}")
        @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = 1000)
        @ResultType(AppUserEntity.class)
        void getUserListBigData(@Param(Constants.WRAPPER) QueryWrapper<AppUserEntity> wrapper, ResultHandler<AppUserEntity> handler);

service.class方法调用

/** * @author yangjunxiong * @date 2020/12/23 10:51 * @param sqlSelect 需要返回的字段 逗号分隔 字段越少查询越快 * @param subDeptIdList * @return java.util.List<com.allianity.common.learning.entity.AppUserEntity> **/
        @Override
        public List<AppUserEntity> getUserListBigData(String sqlSelect,List<Long> subDeptIdList) { 
            List<AppUserEntity> appUserEntities = new ArrayList<>();
            QueryWrapper<AppUserEntity> queryWrapper = new QueryWrapper<>();
            queryWrapper.select(sqlSelect);
            queryWrapper.eq("status", 0);
            queryWrapper.eq("del_flag", 0);
            queryWrapper.in("dept_id", subDeptIdList);
            //流式条件查询
            this.baseMapper.getUserListBigData(queryWrapper, resultContext -> Optional.ofNullable(resultContext.getResultObject())
                    .ifPresent(appUserEntities::add));
            return appUserEntities;
        }

demo2  
批量处理

一、在需要使用流式查询的mapper文件中，定义流式查询方法

package com.unionpay.dao.db2;
    
    import com.baomidou.mybatisplus.core.mapper.BaseMapper;
    import com.unionpay.entity.TblMallOrder;
    import org.apache.ibatis.annotations.*;
    import org.apache.ibatis.mapping.ResultSetType;
    import org.apache.ibatis.session.ResultHandler;
    
    @Mapper
    public interface TblMallOrderDao extends BaseMapper<TblMallOrder> { 
        @Select("${sql}")
        @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = 1000)
        @ResultType(TblMallOrder.class)
        void dynamicSelectLargeData1(@Param("sql") String sql, ResultHandler<TblMallOrder> handler);
    
        @Select("${sql}")
        @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = 1000)
        @ResultType(Map.class)
        void dynamicSelectLargeData2(@Param("sql") String sql, ResultHandler<Map> handler);
    }

二、使用示例

@RestController
    public class TestSearchLargeData { 
        // 这是每批处理的大小
        private final static int BATCH_SIZE = 1000;
        private int size;
        // 存储每批数据的临时容器
        private List<TblMallOrder> mallOrders;
        
    
        @Autowired
        private TblMallOrderDao tblMallOrderDao;
    
        @GetMapping("/getLargeData1")
        public void getLargeData1() { 
            String sql = "select * from t_mall_order";
            tblMallOrderDao.dynamicSelectLargeData1(sql, new ResultHandler<TblMallOrder>() { 
                @Override
                public void handleResult(ResultContext<? extends TblMallOrder> resultContext) { 
                    TblMallOrder tblMallOrder = resultContext.getResultObject();
                    System.out.println(tblMallOrder);
                }
            });
        }
    
        @GetMapping("/getLargeData2")
        public void getLargeData2() { 
            String sql = "select * from t_mall_order";
            tblMallOrderDao.dynamicSelectLargeData1(sql, new ResultHandler<TblMallOrder>() { 
                @Override
                public void handleResult(ResultContext<? extends TblMallOrder> resultContext) { 
                    TblMallOrder tblMallOrder = resultContext.getResultObject();
                    System.out.println(tblMallOrder);
                    // 你可以看自己的项目需要分批进行处理或者单个处理，这里以分批处理为例
                    mallOrders.add(tblMallOrder);
                    size++;
                    if (size == BATCH_SIZE) { 
                        handle();
                    }
                }
            });
            //用来完成最后一批数据处理
            handle();
        }
        /** * 数据处理 */
        private void handle(){ 
            try{ 
                // 在这里可以对你获取到的批量结果数据进行需要的业务处理
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                // 处理完每批数据后后将临时清空
                size = 0;
                mallOrders.clear();
            }
        }
    }

三、总结  
Oracle和DB2，当我们执行一个SQL查询语句的时候，需要在客户端和服务器端都打开一个游标，并且分别申请一块内存空间，作为存放查询的数据的一个缓冲区。这块内存区，存放多少条数据就由fetchsize来决定，同时每次网络包会传送fetchsize条记录到客户端。应该很容易理解，如果fetchsize设置为20，当我们从服务器端查询数据往客户端传送时，每次可以传送20条数据，但是两端分别需要20条数据的内存空闲来保存这些数据。fetchsize决定了每批次可以传输的记录条数，但同时，也决定了内存的大小。这块内存，在oracle服务器端是动态分配的。而在客户端，PS对象会存在一个缓冲中（LRU链表），也就是说，这块内存是事先配好的，应用端内存的分配在conn.prepareStatement(sql)或都conn.CreateStatement(sql)的时候完成。  
MySQL本身并没有FetchSize方法, 它是使用CS阻塞的方式，通过网络流控制服务端，MySQL在执行ResultSet.next()方法时，会通过数据库连接一条一条的返回。MySQL按照自己的节奏不断的把buffer写回网络中。flush buffer的过程是阻塞式的，也就是说如果网络中发生了拥塞，send buffer被填满，会导致buffer一直flush不出去，那MySQL的处理线程会阻塞，从而避免数据把客户端内存撑爆。这样带来的问题：如果使用了流式查询，一个MySQL数据库连接同一时间只能为一个ResultSet对象服务，并且如果该ResultSet对象没有关闭，势必会影响其他查询对数据库连接的使用！此为大坑，所以sharding-sphere费劲心思要提供两种数据库连接模式，如果应用对数据库连接的消耗要求严苛，那么流式查询就不再适合。

package com.aegonthtf.fate.dao.secondary;
    
    import com.aegonthtf.fate.constant.CommonConstant;
    import com.aegonthtf.fate.entity.user.TuDAgent;
    import com.baomidou.mybatisplus.core.mapper.BaseMapper;
    import org.apache.ibatis.annotations.*;
    import org.apache.ibatis.mapping.ResultSetType;
    import org.apache.ibatis.session.ResultHandler;
    
    import java.util.Date;
    
    /** * 营销员维度表(TuDAgent)表数据库访问层 * * @author yangjunxiong * @since 2021-05-11 16:30:00 */
    @Mapper
    public interface TuDAgentDao extends BaseMapper<TuDAgent> { 
    
        //流式批量查询处理数据
    // @Select("select * from tu_d_agent")//测试用
        @Select("select * from tu_d_agent t where AGENT_STATUS = 'Active' and lcd >= #{startTime} and lcd <= #{endTime}")
        @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = CommonConstant.BATCH_SIZE)
        @ResultType(TuDAgent.class)
        void getUserListByLcdBigData(@Param("startTime") Date startTime, @Param("endTime") Date endTime, @Param("handler") ResultHandler<TuDAgent> handler);
    
        //流式批量查询处理数据
    // @Select("select * from tu_d_agent t where lcd <= #{date} LIMIT 10")//测试用
        @Select("select * from tu_d_agent where AGENT_STATUS = 'Active'")
        @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = CommonConstant.BATCH_SIZE)
        @ResultType(TuDAgent.class)
        void getUserListBigData(@Param("handler") ResultHandler<TuDAgent> handler);
    }

package com.aegonthtf.fate.service.user.impl;
    
    import cn.com.common.util.util.StringUtils;
    import com.aegonthtf.fate.constant.CommonConstant;
    import com.aegonthtf.fate.dao.secondary.TuDAgentDao;
    import com.aegonthtf.fate.entity.user.FateAgent;
    import com.aegonthtf.fate.entity.user.FateGoals;
    import com.aegonthtf.fate.entity.user.FateRecord;
    import com.aegonthtf.fate.entity.user.TuDAgent;
    import com.aegonthtf.fate.service.convert.TuDagentToFateAgentConverter;
    import com.aegonthtf.fate.service.convert.TuDagentToFateRecordConverter;
    import com.aegonthtf.fate.service.user.*;
    import com.aegonthtf.fate.util.JodaTimeUtils;
    import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
    import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
    import org.apache.commons.lang3.tuple.ImmutablePair;
    import org.joda.time.LocalDateTime;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    import org.springframework.util.CollectionUtils;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Objects;
    import java.util.Optional;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.Executor;
    
    /** * 营销员维度表(TuDAgent)表服务实现类 * * @author yangjunxiong * @since 2021-05-11 16:30:01 */
    @Service("tuDAgentService")
    public class TuDAgentServiceImpl extends ServiceImpl<TuDAgentDao, TuDAgent> implements TuDAgentService, InitializingBean { 
        @Autowired
        public               FateAgentService      fateAgentService;
        @Autowired
        public               TuDAgentService       tuDAgentService;
        @Autowired
        private              FateBranchService     fateBranchService;
        @Autowired
        private              FateGoalsService      fateGoalsService;
        @Autowired
        private              FateApolloActService  fateApolloActService;
        @Autowired
        private              FateRecordService     fateRecordService;
        @Autowired
        private              AisuattendanceService aisuattendanceService;
        @Autowired
        private              AisuleaveService      aisuleaveService;
        private              int                   size1;//计数器1
        @Autowired
        @Qualifier("tuDAgentServiceImpl1")
        private              Executor              executor1;
        private volatile     BlockingQueue         tuDAgentServiceImplQueue1; // 存储每批数据的临时容器
        private              int                   size2;//计数器2
        @Autowired
        @Qualifier("tuDAgentServiceImpl2")
        private              Executor              executor2;
        private volatile     BlockingQueue         tuDAgentServiceImplQueue2; // 存储每批数据的临时容器
    
        @Override
        public void syncFateAgent(LocalDateTime staratDateTime) { 
            ImmutablePair<LocalDateTime, LocalDateTime> timeDay = JodaTimeUtils.criticalTimeDay(staratDateTime.toLocalDate());
            this.baseMapper.getUserListByLcdBigData(timeDay.getLeft().toDate(), timeDay.getRight().toDate(),
                    resultContext -> { 
                        Optional.ofNullable(resultContext.getResultObject())
                                .ifPresent(this.tuDAgentServiceImplQueue1::add);
                        size1++;
                        if (size1 >= CommonConstant.BATCH_SIZE) { 
                            this.handle1(staratDateTime);
                        }
                    });
            //用来完成最后一批数据处理
            this.handle1(staratDateTime);
        }
    
        /** * 数据处理 */
        @Transactional
        public void handle1(LocalDateTime staratDateTime) { 
            try { 
                // 在这里可以对你获取到的批量结果数据进行需要的业务处理
                List<TuDAgent> tuDAgents1 = new ArrayList<>();
                this.tuDAgentServiceImplQueue1.drainTo(tuDAgents1, CommonConstant.BATCH_SIZE);
                if (CollectionUtils.isEmpty(tuDAgents1)) { 
                    return;
                }
                List<FateAgent> fateAgents = new ArrayList<>();
                String day = staratDateTime.toString("d");
                tuDAgents1.parallelStream().forEach(item -> { 
                    FateAgent fateAgent = this.fateAgentService.getFateAgentByAgentCode(item.getAgentCode());
                    FateAgent to;
                    if (Objects.isNull(fateAgent)) { 
                        to = TuDagentToFateAgentConverter.to(item, staratDateTime, new FateAgent());
                    } else { 
    // to = TuDagentToFateAgentConverter.to(item, fateAgent);
                        //有记录的更新AGENT_STATUS
                        to = fateAgent;
                        Optional.ofNullable(item.getAgentStatus()).ifPresent(fateAgent::setAgentStatus);
                    }
                    if (Objects.isNull(to)) { 
                        return;
                    }
                    //处理签约事件
                    this.fateApolloActService.contractEventHandling(staratDateTime, item);
                    //获取省一 二 三
                    if (StringUtils.isNotBlank(to.getBranchcode())) { 
                        FateAgent fateAgentOfBranch123 = this.fateBranchService.getFateBranchByBranchCode(to.getBranchcode());
                        if (!Objects.isNull(fateAgentOfBranch123)) { 
                            Optional.ofNullable(fateAgentOfBranch123.getBrhFst()).ifPresent(to::setBrhFst);
                            Optional.ofNullable(fateAgentOfBranch123.getBrhScd()).ifPresent(to::setBrhScd);
                            Optional.ofNullable(fateAgentOfBranch123.getBrhThd()).ifPresent(to::setBrhThd);
                        }
                    }
                    //获取目标APE 每月一号 二号更新 GAPE和目标新增人力gManpower
                    if (StringUtils.equals(day, "1") || StringUtils.equals(day, "2")) { 
                        FateGoals one = this.fateGoalsService.getOne(new QueryWrapper<FateGoals>()
                                .lambda()
                                .eq(FateGoals::getAgentCode, to.getAgentCode())
                                .last(CommonConstant.LIMIT1)
                        );
                        Optional.ofNullable(one).ifPresent(itemFateGoals -> { 
                            //目标达成-APE
                            to.setGApe(Objects.isNull(itemFateGoals.getGApe()) ? 0D : one.getGApe());
                            //目标-新增人力
                            to.setGManpower(Objects.isNull(itemFateGoals.getGManpower()) ? 0 : one.getGManpower());
                        });
                    }
                    //达成-新增人力 aManpower
                    Integer aManpower = this.tuDAgentService.getFateAgentByAManpowerAndMonth(to.getAgentCode(), staratDateTime);
                    to.setAManpower(Objects.isNull(aManpower) ? 0 : aManpower);
                    fateAgents.add(to);
                });
                this.fateAgentService.saveOrUpDateBatchEntity(fateAgents);
    
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                // 处理完每批数据后后将临时清空
                size1 = 0;
            }
        }
    
        @Override
        public void initFateRecord(LocalDateTime staratDateTime) { 
            this.baseMapper.getUserListBigData(resultContext -> { 
                Optional.ofNullable(resultContext.getResultObject())
                        .ifPresent(this.tuDAgentServiceImplQueue2::add);
                size2++;
                if (size2 >= CommonConstant.BATCH_SIZE) { 
                    this.handle2(staratDateTime);
                }
            });
            //用来完成最后一批数据处理
            this.handle2(staratDateTime);
        }
    
        /** * 数据处理 */
        @Transactional
        public void handle2(LocalDateTime staratDateTime) { 
            try { 
                // 在这里可以对你获取到的批量结果数据进行需要的业务处理
                List<TuDAgent> tuDAgents2 = new ArrayList<>();
                this.tuDAgentServiceImplQueue1.drainTo(tuDAgents2, CommonConstant.BATCH_SIZE);
                if (CollectionUtils.isEmpty(tuDAgents2)) { 
                    return;
                }
                List<FateRecord> fateRecords = new ArrayList<>();
                tuDAgents2.parallelStream().forEach(item -> { 
                    if (Objects.isNull(item)) { 
                        return;
                    }
                    FateRecord fateRecord = this.fateRecordService.getTodayByAgentCode(item.getAgentCode(), staratDateTime);
                    FateRecord to;
                    if (Objects.isNull(fateRecord)) { 
                        to = TuDagentToFateRecordConverter.to(item, staratDateTime, new FateRecord());
                    } else { 
                        to = TuDagentToFateRecordConverter.to(item, staratDateTime, fateRecord);
                    }
                    if (Objects.isNull(to)) { 
                        return;
                    }
                    //获取省一 二 三
                    if (StringUtils.isNotBlank(to.getBranchcode())) { 
                        FateAgent fateAgentOfBranch123 = this.fateBranchService.getFateBranchByBranchCode(to.getBranchcode());
                        if (!Objects.isNull(fateAgentOfBranch123)) { 
                            Optional.ofNullable(fateAgentOfBranch123.getBrhFst()).ifPresent(to::setBrhFst);
                            Optional.ofNullable(fateAgentOfBranch123.getBrhScd()).ifPresent(to::setBrhScd);
                            Optional.ofNullable(fateAgentOfBranch123.getBrhThd()).ifPresent(to::setBrhThd);
                        }
                    }
                    //判断工作日逻辑
                    //1.根据代理人工号查询aisuattendance表有无当天记录。
                    //SELECT COUNT(1) FROM AISUATTENDANCE T WHERE T.AI_REC_ENDDT IS NULL AND T.AI_AGID = '工号' AND T.AI_ATTENDANCE_DATE = '日期';
                    //2.若无记录，则当天非工作日，若有记录，查当前代理人是否请假。
                    //select count(1) from aisuleave where ai_agid='0000223106' and ai_leave_enddt >='20201230' and ai_leave_stdt <='20201230';
                    //3.若仍无记录，则当天是工作日。
                    String workFlag;
                    Integer count1 = this.aisuattendanceService.getWorkFlagByAgentCode(to.getAgentCode(), staratDateTime);
                    if (count1 <= 0) { 
                        workFlag = "N";
                    } else { 
                        Integer count2 = this.aisuleaveService.getWorkFlagByAgentCode(to.getAgentCode(), staratDateTime);
                        workFlag = (count2 <= 0) ? "Y" : "N";
                    }
                    to.setWorkFlag(workFlag); //是否是工作日(Y:N) 每天7点批处理更新
                    Integer count = this.fateRecordService.getYesterdayFateRecordByAgentCode(to.getAgentCode());
                    to.setRecordDays(count); //连续打卡天数
                    fateRecords.add(to);
                });
                this.fateRecordService.saveOrUpDateBatchEntity(fateRecords);
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                // 处理完每批数据后后将临时清空
                size2 = 0;
            }
        }
    
        //查询该工号当月 //当月达成-新增人力 aManpower
        @Override
        public Integer getFateAgentByAManpowerAndMonth(String agentCode, LocalDateTime staratDateTime) { 
            ImmutablePair<LocalDateTime, LocalDateTime> timeMonth = JodaTimeUtils.criticalTimeMonth(staratDateTime);
            return this.count(new QueryWrapper<TuDAgent>().lambda()
                    .eq(TuDAgent::getRefferCode, agentCode)
                    .ge(TuDAgent::getFcd, timeMonth.left.toDate())
                    .le(TuDAgent::getFcd, timeMonth.right.toDate())
            );
        }
    
        @Override
        public void afterPropertiesSet() { 
            ThreadPoolTaskExecutor poolTaskExecutor1 = (ThreadPoolTaskExecutor) this.executor1;
            ThreadPoolTaskExecutor poolTaskExecutor2 = (ThreadPoolTaskExecutor) this.executor2;
            this.tuDAgentServiceImplQueue1 = poolTaskExecutor1.getThreadPoolExecutor().getQueue();
            this.tuDAgentServiceImplQueue2 = poolTaskExecutor2.getThreadPoolExecutor().getQueue();
        }
    }