springboot 整合 mongodb实现 批量更新数据

现需求：需要批量将1000个数据先查询在更新到mongodb（如果查询不到数据，则添加数据）

1：工具类BathUpdateOptions

import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;

public class BathUpdateOptions {
    
    private Query query;
    private Update update;
    private boolean upsert = true;
    private boolean multi = false;

    public Query getQuery() {
        return query;
    }

    public void setQuery(Query query) {
        this.query = query;
    }

    public Update getUpdate() {
        return update;
    }

    public void setUpdate(Update update) {
        this.update = update;
    }

    public boolean isUpsert() {
        return upsert;
    }

    public void setUpsert(boolean upsert) {
        this.upsert = upsert;
    }

    public boolean isMulti() {
        return multi;
    }

    public void setMulti(boolean multi) {
        this.multi = multi;
    }

}

2：需要更新的domain

@Document(collection = "video_show_view")
public class VideoShowView {
    //唯一值
    private String id;
    //节目id
    private String cid;
    //app播放次数
    private String view;
    //app虚拟播放次数
    private String virtualViews;
    //最后更新时间 时间戳
    private String updateTime;
　　//get set ......
}

3：获取BathUpdateOptions 集合

/**
 * @author xuyou
 * @Package com.migu.live.mao
 * @Description:
 * @date 2018/6/11 16:13
 */
@Repository
public class VideoShowViewMao {
    @Autowired
    private MongoTemplate mongoTemplate;
    
    public BathUpdateOptions getBathUpdateOptions(VideoShowView videoShowView){
        BathUpdateOptions options = new BathUpdateOptions();
        Query query = new Query();
        //查询条件
        query.addCriteria(Criteria.where("cid").is(videoShowView.getCid()));
        query.addCriteria(Criteria.where("types").is(videoShowView.getTypes()));
        options.setQuery(query);
        //mongodb 默认是false,只更新找到的第一条记录，如果这个参数为true,就把按条件查出来多条记录全部更新。
        options.setMulti(true);
        Update update = new Update();
        //更新内容
        update.set("view", videoShowView.getView());
        update.set("updateTime", videoShowView.getUpdateTime());
        options.setUpdate(update);
        return options;
    }
    
    public void bathUpdate(List<BathUpdateOptions> bups){
        BathUpdateUtil.bathUpdate(mongoTemplate, VideoShowView.class, bups);
    }
}

4：操作mongodb的工具类BathUpdateUtil

import java.util.ArrayList;
import java.util.List;

import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.mapping.Document;

import com.migu.live.data.BathUpdateOptions;
import com.mongodb.BasicDBObject;
import com.mongodb.CommandResult;
import com.mongodb.DBCollection;
import com.mongodb.DBObject;

public class BathUpdateUtil {

    /**
     * @description:批量更新mongodb数据
     * @author: xuyou
     * @date: 2018年8月14日 上午11:28:29 
     */
    public static int bathUpdate(MongoTemplate mongoTemplate, Class<?> entityClass, 
        List<BathUpdateOptions> options) {
        String collectionName = determineCollectionName(entityClass);
        return doBathUpdate(mongoTemplate.getCollection(collectionName), 
             collectionName, options, true);
    }
    
    private static String determineCollectionName(Class<?> entityClass) {
        if (entityClass == null) {
            throw new InvalidDataAccessApiUsageException(
                    "No class parameter provided, entity collection can't be determined!");
        }
        String collName = entityClass.getSimpleName();
        if(entityClass.isAnnotationPresent(Document.class)) {
            Document document = entityClass.getAnnotation(Document.class);
            collName = document.collection();
        } else {
            collName = collName.replaceFirst(collName.substring(0, 1)
                      ,collName.substring(0, 1).toLowerCase()) ;
        }
        return collName;
    }

    private static int doBathUpdate(DBCollection dbCollection, String collName,         
                       List<BathUpdateOptions> options, boolean ordered) {
        DBObject command = new BasicDBObject();
        command.put("update", collName);
        List<BasicDBObject> updateList = new ArrayList<BasicDBObject>();
        for (BathUpdateOptions option : options) {
            BasicDBObject update = new BasicDBObject();
            update.put("q", option.getQuery().getQueryObject());
            update.put("u", option.getUpdate().getUpdateObject());
            update.put("upsert", option.isUpsert());
            update.put("multi", option.isMulti());
            updateList.add(update);
        }
        command.put("updates", updateList);
        command.put("ordered", ordered);
        CommandResult commandResult = dbCollection.getDB().command(command);
        return Integer.parseInt(commandResult.get("n").toString());
    }
}

5：业务代码  可根据实际需求 进行修改

/**
     * @description:执行更新
     * @param liveKeys 需要更新的集合
     * @return void     
     * @author: xuyou
     * @date: 2018年8月14日 上午11:44:06 
     * @throws
     */
    public void bathUpdateMongoDB (Set<String> liveKeys){
        //将直播播放次数入mongoDB
        List<BathUpdateOptions> bupsList = new ArrayList<BathUpdateOptions>();
        for (String id : liveKeys) {
            VideoShowView videoShowView = new VideoShowView();
            //设置一些更新条件 此处省略
            videoShowView.setUpdateTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
            BathUpdateOptions options = videoShowViewMao.getBathUpdateOptions(videoShowView);
            bupsList.add(options);
            if(bupsList.size() >= 1000){
                logger.info("bupsList : {}",bupsList.size());
                videoShowViewMao.bathUpdate(bupsList);
                bupsList = new ArrayList<BathUpdateOptions>();
            }
        }
        //TODO 更新liveList剩余少于1000的数据
        logger.info("bupsList : {}",bupsList.size());
        if (bupsList != null && bupsList.size() > 0) {
            videoShowViewMao.bathUpdate(bupsList);
        }
    }

6;pom文件

    <!-- 对mongodb的支持 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb</artifactId>
            <version>2.0.2.RELEASE</version>
        </dependency>

7：配置

spring.data.mongodb.uri=mongodb://10.200.85.97:27017,10.200.85.98:27017,10.200.85.99:27017/data_consumer