问题描述

? ? ? ?有一个分类预测的接口，主要业务逻辑是输入一段文本，接口内部调用模型对文本进行分类预测。模型数据是直接在内存中，所以预测的过程本身很快。预测完成之后，往预测记录表插入一条数据。后续有其他应用会对该记录进行矫正，判断是否预测成功，以便后续进行自学习。

? ? ? ?接口上线初期好评如潮，不管是做智能派遣还是经办单位流转预测使用效果都很好，随着自学习功能加入，预期应用应该会越跑越好。但是前两天客户现场突然反馈系统很慢，做智能派遣和流转的时候要三四秒甚至五六秒才能出结果。

? ? ? ?最开始想的是现场的分类模型太多，模型树也比较深，所以一次多级预测不可避免会比较耗时，但是核查后发现最多的模型树只有四级，单个模型直接验证预测结果都是毫秒出结果，但是直接调用接口就很慢。如果不是算法模型的问题，也不是机器资源的问题，那一定是代码逻辑中有没有考虑到的点，而且随着数据量激增，问题越来越明显。

? ? ? ?经过简单排查，发现问题非常低级，造成慢的主要原因是因为预测记录表的数据已经将近100w了，而且这个表没有任何索引，只有一个自增主键，程序中，因为业务逻辑需要，每次调用预测接口都会有一个查询再更新或者插入的动作，并且是同步的。那么随着数据量的激增，接口的RT注定会越来越慢。

解决方案

? ? ? ?问题是个小问题，主要原因还在于接口设计的初期没有做深入的考虑，代码写的很漂亮，只可惜不经考验，绣花枕头一个。对于这种问题，最直观的方案就是解耦，预测和预测记录入库做成异步的。当然如果有业务场景硬要求不得不做成同步，也可以从代码层面和数据库层面做优化增加处理速度。

1 消息中间件

（1）可以直接引入kafka或者RocketMQ进行解耦，这种比较保险，数据不会丢，而且可以做到核心功能和非核心功能解耦。是主流解决方案，并且支持横向扩展和分布式。

（2）如果不想使用MQ，也可以直接使用Redis，Redis也可以实现消息队列的功能，参考这里

2 代码及数据库优化

? ? ? ?如果不得不同步操作，那就要从代码层面进行优化和调整，主要以下几个思路：

（1）给查询的字段添加索引，通过数据库的能力提升IO

（2）设置分区表或者天表（月表）的概念，降低单表的数据量

（3）代码优化，将查询后的更新或者插入合并为一个sql操作，参考这里

insertOrUpdate的实现是基于mysql的on duplicate key update 来实现的。

使用ON DUPLICATE KEY UPDATE，如果行作为新行插入，则每行受影响的行值为1。如果更新现有行，则每行受影响的行值为2；如果将现有行设置为其当前值，则每行受影响的行值为0（可以通过配置，使其受影响的行值为1）。

官方地址：

13.2.6.2 INSERT … ON DUPLICATE KEY UPDATE Statement

? ? ? ?对于异步解耦这种方案，如果之前应用本身没有MQ、Redis，为了解决该问题安装这些中间件，从某种程度上来说这无疑增加了系统复杂度好运维集成的工作量。基于此，我们也可以使用java自身的多线程来实现异步，基本思路就是在需要进行IO操作的地方，直接开启一个新线程去处理。

? ? ? ?但是如果请求量很大，这无疑会造成频繁的线程创建、释放资源问题，如果引入线程池，又可能会出现因为资源用完阻塞的情况，这不能根本上解决同步的问题。?

? ? ? ?可以通过java多线程模拟消息队列，在需要进行IO操作的业务代码处，将业务数据封装为Bo放到一个队列中。有一个独立的线程对队列进行消费处理即可。可以参考这里

? ? ? ?我主要是使用了ConcurrentLinkedQueue来解决了此问题，基本思路是，创建了一个定时任务，每30秒执行一次，每次都去处理ConcurrentLinkedQueue队列中的数据，将数据入库。而业务代码中是将业务数据封装成Bo放到队列中去了。

3?ConcurrentLinkedQueue方案

定时任务

@Component
public class TaskJob
{
    private static final Logger logger = LoggerFactory.getLogger(TaskJob.class);

    @Scheduled(cron = "*/10 * * * * ?") //每10秒执行一次，异步处理预测结果信息，入库
    public void execuPredictResult() throws Exception
    {
        SyncSavePredictResultService syncSavePredictResultService = new SyncSavePredictResultService();
        syncSavePredictResultService.consumeData();
    }
}

异步处理

public class SyncSavePredictResultService
{
    private static final Logger logger = LoggerFactory.getLogger(SyncSavePredictResultService.class);

    //预测记录的消息队列，异步处理入库操作
    public static final ConcurrentLinkedQueue<PredictResultBo> RESULT_BO = new ConcurrentLinkedQueue<>();

    public void consumeData()
    {
        PredictResultBo resultBo = RESULT_BO.poll();
        while (resultBo != null)
        {
            //进行业务逻辑处理
            excuSubPdResult(resultBo.getContent(),
                    resultBo.getBegin(),
                    resultBo.getEnd(),
                    resultBo.getmId(),
                    resultBo.getFlagValue(),
                    resultBo.getPredictResult(),
                    resultBo.getRootPid(),
                    resultBo.getD());
            logger.info("异步写入预测记录，对象内容为:{}", resultBo.toString());
            //更新resultBo
            resultBo = RESULT_BO.poll();
        }
    }

    private void excuSubPdResult(String pstr, long begin, long end, Integer mId, String flagValue, String result, Integer rootPid, Date d)
    {
        NlpSubPredictResults nResult = NlpSubPredictResults.GetInstance().findFirst("select * from nlp_sub_predict_results where flag_value=?", flagValue);
        if (nResult != null)
        {
            nResult.set("predict_result", result)
                   .set("start_time", begin)
                   .set("end_time", end)
                   .set("content", pstr)
                   .set("updated_at", d)
                   .update();
        }
        else
        {
            nResult = NlpSubPredictResults.GetInstance();
            nResult.set("flag_value", flagValue)
                   .set("root_p_id", rootPid)
                   .set("m_id", mId)
                   .set("predict_result", result)
                   .set("start_time", begin)
                   .set("end_time", end)
                   .set("content", pstr)
                   .set("created_at", d)
                   .set("updated_at", d)
                   .save();
        }
    }
}

业务流程

private void excuSubPdResult(String pstr,long begin,long end,Integer mId,String flagValue,String typeId,String typeName,Integer rootPid){
        Map<String, String> rMap = new HashMap<>();
        rMap.put("name",typeName);
        rMap.put("id", typeId);
        String result = JSON.toJSONString(rMap);


        Date                 d       = new Date();
        PredictResultBo predictResultBo = new PredictResultBo(mId,rootPid,flagValue,result,begin,end,pstr,d);
        SyncSavePredictResultService.RESULT_BO.add(predictResultBo);

        /*
        //2022-05-30改为异步
        NlpSubPredictResults nResult = NlpSubPredictResults.GetInstance().findFirst("select * from nlp_sub_predict_results where m_id=? and flag_value=? and root_p_id=?",mId,flagValue,rootPid);
        if(nResult!=null){
            nResult.set("predict_result", result)
                   .set("start_time",begin)
                   .set("end_time",end)
                   .set("content",pstr)
                   .set("updated_at",d)
                   .update();
        }else{
            nResult = NlpSubPredictResults.GetInstance();
            nResult.set("flag_value",flagValue)
                   .set("root_p_id",rootPid)
                   .set("m_id",mId)
                   .set("predict_result", result)
                   .set("start_time",begin)
                   .set("end_time",end)
                   .set("content",pstr)
                   .set("created_at",d)
                   .set("updated_at",d)
                   .save();
        }
         */
    }