码迷,mamicode.com
首页 > 数据库 > 详细

flowable 集成mongodb

时间:2019-01-17 12:07:17      阅读:1256      评论:0      收藏:0      [点我收藏+]

标签:rect   compile   sda   msn   数据库查询   lte   审批   ctas   with   

学无止境,活到老学到老,每天都问自己进步了吗?

 

第一:背景

由于公司每天有至少1500个表单发起,处理待办任务至少7000个,累计历史任务数据已经达到200多w条,时间一长,通过数据库查询已办的任何和我发起的流程巨慢

所以我们考虑到这些数据能不能放入ES或者是mongodb中

流程中心1.0版本集成的是ES,速度确实非常快,提升查询性能近万倍,但是由于ES是一个全文检索的系统,对我们这些业务数据来说,不是很适合,比方说

我们的表单数据,他直接给分词了,不符合业务的要求。

流程中心2.0版本我们就改成mongodb,速度一样达到es的查询效果,解决了以前分词的问题,而且数据结构化查询也非常方便。

第二:集成mongodb策略

集成mongodb的策略有两种

1.利用flowable提供的mongodb的插件来集成,具体可以参考他们提供的demo

2.通过流程实例id和任务id查询表数据,做加工处理,然后在通过消息队列的方式同步到mongodb中

第一种方式就相当于把历史表的数据全部搬到mongodb中,在关系型数据库中不存放任何历史数据,正是由于中原因,我担心历史数据的丢失,所以我采用的是

第二种策略,关系数据库中有一份数据,在mongodb中我也有一份加工后的数据,以保证万无一失。

第三:具体实现

3.1 同步数据

@Component
@RabbitListener(queues = FlowConstant.FLOWABLE_HISTORY_DATAS, containerFactory = "rabbitListenerContainerFactory")
public class HisDataSyncReceiver {
    private static Logger logger = LoggerFactory.getLogger(HisDataSyncReceiver.class);

    @Autowired
    private ISyncHisService syncHisService;

    @RabbitHandler
    public void process(@Payload String message) {

        if (StringUtils.isNotBlank(message)) {
            logger.info("接受到的数据为:" + message);
            try {
                Thread.sleep(100);
                SynHisDataVo data = (SynHisDataVo) JsonUtils.jsonToObj(message, SynHisDataVo.class);
                String type = data.getType();
                String id = data.getId();
                String processStaus = data.getProcessStatus();
                if (type.equals(SynHisDataEnum.EXECUTION.getSn())) {
                    syncHisService.syncExecution(id,processStaus);
                } else if (type.equals(SynHisDataEnum.TASK.getSn())) {
                    syncHisService.syncTask(id,processStaus);
                } else {
                    logger.error("历史队列中错误的数据" + message, message);
                }
            } catch (Exception e) {
                logger.error("历史队列中错误的数据"+ message, e);
            }
        }
    }
}

3.2 查询mongodb的任务数据

public PagerModel<SearchTaskVo> querySearchTaskVoPagerModel(QueryTaskVo params, Query query) {
        org.springframework.data.mongodb.core.query.Query queryParams = new org.springframework.data.mongodb.core.query.Query();
        queryParams.addCriteria(Criteria.where("endTime").exists(true));
        if (StringUtils.isNotBlank(params.getAssignee())) {
            queryParams.addCriteria(Criteria.where("assignee").is(params.getAssignee().trim()));
        }
        if (StringUtils.isNotBlank(params.getCreator())) {
            queryParams.addCriteria(Criteria.where("creator").is(params.getCreator().trim()));
        }
        if (StringUtils.isNotBlank(params.getCreateName())) {
            queryParams.addCriteria(Criteria.where("creatorName").is(params.getCreateName().trim()));
        }
        if (StringUtils.isNotBlank(params.getExecutionName())) {
            Pattern pattern = Pattern.compile("^.*" + params.getExecutionName().trim() + ".*$", Pattern.CASE_INSENSITIVE);
            //通过流程标题或流程编号复合查询
            Criteria criteriaNameOrKey = new Criteria();
            criteriaNameOrKey.orOperator(Criteria.where("procInstName").regex(pattern),Criteria.where("businessKey").regex(pattern),Criteria.where("creatorName").regex(pattern));
            queryParams.addCriteria(criteriaNameOrKey);
        }
        if (params.getProcessStatus() != null) {
            queryParams.addCriteria(Criteria.where("processStatus").is(params.getProcessStatus().trim()));
        }
        if (StringUtils.isNotBlank(params.getStartTime()) && StringUtils.isNotBlank(params.getEndTime())) {
            queryParams.addCriteria(Criteria.where("startTime").gte(params.getStartTime().trim()).lte(params.getEndTime().trim()));
        }
        if (StringUtils.isNotBlank(params.getSystemSn())) {
            queryParams.addCriteria(Criteria.where("systemSn").is(params.getSystemSn().trim()));
        }
        queryParams.with(new Sort(Sort.Direction.DESC, "endTime"));
        int i = query.getInitPageIndex() < 1 ? 1 : query.getInitPageIndex();
        queryParams.skip((i - 1) * query.getPageSize()).limit(query.getPageSize());
        List<SearchTaskVo> datas = mongoTemplate.find(queryParams, SearchTaskVo.class, SearchTaskVo.TASK_TABLE);
        long count = mongoTemplate.count(queryParams, SearchTaskVo.TASK_TABLE);
        //去重
        datas = this.removal(datas);
        //翻译转阅
        this.turnReadTaskData(datas);
        //排序
        if (MapUtils.isNotEmpty(params.getOrderbyMap())){
            this.taskCommonSort(params.getOrderbyMap(),datas);
        }
        PagerModel<SearchTaskVo> pm = new PagerModel<>(count, datas);
        return pm;
    }

3.3.查询mongodb的实例数据

public PagerModel<SearchExecutionVo> querySearchExecutionVoPagerModel(QueryTaskVo params, com.dragon.tools.pager.Query query) {
        Query queryParams = new Query();
        //创建者工号
        if (StringUtils.isNotBlank(params.getCreator())) {
            queryParams.addCriteria(Criteria.where("startUser").is(params.getCreator()));
        }
        //创建者姓名
        if (StringUtils.isNotBlank(params.getCreateName())) {
            queryParams.addCriteria(Criteria.where("creatorName").is(params.getCreateName()));
        }
        if (StringUtils.isNotBlank(params.getProcessDefinitionKey())) {
            queryParams.addCriteria(Criteria.where("proDefKey").is(params.getProcessDefinitionKey()));
        }else {
            //转阅流程在已发不让显示
            queryParams.addCriteria(Criteria.where("proDefKey").ne(FlowConstant.TURN_READ));
        }
        if (StringUtils.isNotBlank(params.getExecutionName())) {
            Pattern pattern = Pattern.compile("^.*" + params.getExecutionName() + ".*$", Pattern.CASE_INSENSITIVE);
           // queryParams.addCriteria(Criteria.where("name").regex(pattern));
            // queryParams.addCriteria(Criteria.where("businessKey").regex(pattern));
            //通过流程标题或流程编号复合查询
            Criteria criteriaNameOrKey= new Criteria();
            criteriaNameOrKey.orOperator(Criteria.where("name").regex(pattern),
                    Criteria.where("businessKey").regex(pattern));
            queryParams.addCriteria(criteriaNameOrKey);


        }
        if (StringUtils.isNotBlank(params.getProcessStatus())) {
            queryParams.addCriteria(Criteria.where("processStatus").is(params.getProcessStatus().trim()));
        }
        if (StringUtils.isNotBlank(params.getStartTime()) && StringUtils.isNotBlank(params.getEndTime())) {
            queryParams.addCriteria(Criteria.where("startTime").gte(params.getStartTime().trim()).lte(params.getEndTime().trim()));
        }
        if (StringUtils.isNotBlank(params.getSystemSn())) {
            queryParams.addCriteria(Criteria.where("systemSn").is(params.getSystemSn().trim()));
        }
        //默认开始时间降序
        queryParams.with(new Sort(Sort.Direction.DESC, "startTime"));
        queryParams.skip(((query.getInitPageIndex()==0?1:query.getInitPageIndex())-1)*query.getPageSize()).limit(query.getPageSize());
        List<SearchExecutionVo> datas = mongoTemplate.find(queryParams, SearchExecutionVo.class, SearchExecutionVo.EXECUTION_TABLE);
        //当前审批人查询出来
        this.createApprover(datas);
        //转阅流程特殊处理
        this.turnReadExecutionData(datas);
        if (MapUtils.isNotEmpty(params.getOrderbyMap())){
            this.excutionCommonSort(params.getOrderbyMap(),datas);
        }
        long count = mongoTemplate.count(queryParams, SearchExecutionVo.class, SearchExecutionVo.EXECUTION_TABLE);
        return new PagerModel<>(count, datas);
    }

 

flowable 集成mongodb

标签:rect   compile   sda   msn   数据库查询   lte   审批   ctas   with   

原文地址:https://www.cnblogs.com/liuwenjun/p/10281439.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!