mongodb第二类丢失更新问题

Question

mongodb第二类丢失更新问题

薛冰

515

发布于
2018-10-05

更新于
2018-10-05

新手上路，请多包涵

问题描述

我想对mongodb数据库进行更新操作,将一个删除了部分元素的数组存进mongodb,但是当有至少两个线程同时执行这一任务时,就出现了第二类更新丢失问题,比如原来的数组是[0,1],线程A删除了0,将数组存进数据库,线程B删除了1,将数组存进数据库,我预想的结果是数组应该为空,但是结果是[1]

问题出现的环境背景及自己尝试过哪些方法

我是一名大二的学生,对数据库和并发的了解很有限,通过google才了解(或许了解了)了导致这一现象的原因,根据我所获得的知识,我想到四种解决办法(并不确定能解决)
一. 给文档加读锁,在读取-修改这一操作完成之前不让其他线程读取
二. 使用乐观锁,如果发现数据不是最新的就重试
三. 使用redis,redis似乎不会出现这样的问题
四. 将取出的FileMsg对象放在内存中,并设为线程安全的,在对这个对象进行修改
但是我并不知道怎样去实现第一,二个方案,也不确定哪种方案最好,更不知道有没有更好的解决方案
我尝试使用乐观锁,在pojo中加上了@Version注解和version字段,但是我不知道如何重试(网上的代码看不懂...)

相关代码

    // 自定义的更新方法
    @Autowired
    protected MongoTemplate mongoTemplate;
    public void updateUnfinishedById(String id, LinkedList<Integer> list){
        Query query = new Query(Criteria.where("_id").is(id));
        Update update = new Update().set("unfinishedChunk",list);
        mongoTemplate.findAndModify(query,update,FileMsg.class);
        //这样似乎还是无法避免第二类丢失更新,第二个线程还是读取了第一个线程修改前的数据
    }


//service接口的实现代码
import com.xuebling.newpoetryspread.common.utils.UploadUtils;
import com.xuebling.newpoetryspread.dao.FileMsgRepository;
import com.xuebling.newpoetryspread.pojo.FileChunk;
import com.xuebling.newpoetryspread.pojo.FileMsg;
import com.xuebling.newpoetryspread.pojo.enums.ResponseMsg;
import com.xuebling.newpoetryspread.pojo.result.Response;
import com.xuebling.newpoetryspread.pojo.result.ResponseData;
import com.xuebling.newpoetryspread.service.UploadFileService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;

import java.io.File;
import java.io.IOException;
import java.util.LinkedList;
import java.util.Optional;

@Service
public class UploadFileServiceImpl implements UploadFileService {
    @Autowired
    private FileMsgRepository fileMsgRepository;
    protected Logger logger =  LoggerFactory.getLogger(this.getClass());

    @Value("${fileStorePath}")
    private String fileStorePath;

    //对文件块进行md5校验,返回处理结果,如果所有文件块都上传完毕,则进行源文件完整性校验
    @Transactional(isolation = Isolation.SERIALIZABLE)
    @Override
    public Object finishChunkUpload(FileChunk fileChunk) {
        //查找对象
        Optional<FileMsg> fileMsg = fileMsgRepository.findById(fileChunk.getTaskId());
//        logger.error("version为"+fileMsg.get().getVersion());
        //空检查
        if(!fileMsg.isPresent()){
            logger.error("对象不存在,该任务在数据库中没有存档");
            return new Response(ResponseMsg.TARGETNULL);
        }
        //初始化,对前端传的数据进行处理
        logger.info("chunkSize为"+fileMsg.get().getChunkSize());
        fileChunk.init(fileMsg.get().getChunkSize());
        try {
            logger.info("完整文件md5值为"+fileMsg.get().getSourceFileMD5());

            //如果此块文件确实没有被上传
            if(fileMsg.get().getUnfinishedChunk().contains(fileChunk.getChunk())){
                //上传成功将会返回这个文件名
                logger.info("fileStorePath为"+fileStorePath);
                String targetFileName = fileMsg.get().getBeginTime()+fileMsg.get().getFileName();//
                File targetFile = new File(fileStorePath+targetFileName);
                logger.info("目标文件路径为"+targetFile.getAbsolutePath());
                logger.info("目标文件名为"+targetFile.getName());

                //文件不存在则创建新文件
                if(!targetFile.exists()){
                    logger.info("创建了新文件");
                    targetFile.createNewFile();
                }

                logger.info("文件正在写入");
                //验证成功,将文件写入,如果写入失败,则直接报错了
                UploadUtils.writeByRandomAccess(fileChunk,targetFile);
                logger.info("文件写入成功,正在将记录写入数据库");

                //写入成功后,记录下进度
                LinkedList<Integer> linkedList = fileMsg.get().getUnfinishedChunk();
                logger.info("当前的chunk为"+fileChunk.getChunk());
                linkedList.remove(fileChunk.getChunk());
                logger.info("unfishedChunk为"+fileMsg.get().getUnfinishedChunk());
                fileMsgRepository.updateUnfinishedById(fileMsg.get().getTaskId(),linkedList);//fixme 如何解决事务性的问题

                //全部记录被删除或者文件没有被分块,开始校验完整文件的md5值
                if(linkedList.size()==0||fileMsg.get().getChunkNum()==1){
                    logger.info("所有文件块上传完成,正在校验完整文件");
                    String completeFileMD5 = UploadUtils.validateFile(targetFile);
                    logger.info("合并后文件的md5为"+completeFileMD5);
                    if (completeFileMD5.equals(fileMsg.get().getSourceFileMD5())){
                        //返回一个uri
                        logger.info("校验完整性成功");
                        return new ResponseData(ResponseMsg.ALLDONE,targetFileName);
                    }
                    //校验失败
                    else return new ResponseData(ResponseMsg.ALLFAIL,fileMsg);
                }
            }
            else {
                return new Response(ResponseMsg.TARGETNULL);
            }
        } catch (IOException e) {
            logger.error("文件写入失败");
            e.printStackTrace();
        }
        return new Response(ResponseMsg.ONEDONE);
    }
}