mongodb 单事务内使用多线程加速的简单例子

Posted 2022-11-02 李昊轩的博客

事务内 如果有重复性比较强且线程安全的重复性循环操作.
可以考虑使用并行来替代循环

下面是java 操作 mongo 实现单事务内使用多线程的例子

package com.mindsuite.saas.ztest;

import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.client.ClientSession;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

@RestController
@RequestMapping("/test6")
public class test666 

    @GetMapping("/6")
    public Object testTodo() 

        //创建并发控制器
        int total = 5;
        CountDownLatch countDownLatch = new CountDownLatch(total);


        //获取线程池执行的实例（需要经过TtlExecutors修饰一下）
        Executor executor = Executors.newFixedThreadPool(total);

        //获取ClientSession
        MongoClient mongoClient = new MongoClient(new MongoClientURI("mongodb://192.168.0.177:40000,192.168.0.177:40001,192.168.0.177:40002/test666?slaveOk=true&write=1&readPreference=secondaryPreferred"));
        ClientSession clientSession = mongoClient.startSession();
        MongoTemplate mainTemplate = new MongoTemplate(mongoClient, "test666").withSession(clientSession);


        try 
            //开启事务
            clientSession.startTransaction();


            HashMap<String, Object> tenant = new HashMap<>();
            tenant.put("name", "ksl");
            tenant.put("money", 1);

            mainTemplate.save(tenant, "tenant");

            for (int i = 0; i < total; i++) 

                //开启一个线程，执行任务处理
                executor.execute(() -> 
                    MongoTemplate subMongoTemplate = SelfMongoTemplate.getMongoDBTemplate("666").withSession(clientSession);

                    HashMap<String, Object> tenant2 = new HashMap<>();
                    tenant2.put("name", "ksl2");
                    tenant2.put("money", System.currentTimeMillis());

                    subMongoTemplate.save(tenant2, "tenant");


//
//                Update update1 = new Update();
//                update1.set("total_goods_num", rd.nextInt(100));
//
//                subMongoTemplate.updateFirst(query, update1, "bill");

                    countDownLatch.countDown();
                );
            
            countDownLatch.await();

            //提交事务
            clientSession.commitTransaction();
         catch (Exception e) 
            //撤销事务
            clientSession.abortTransaction();

            e.printStackTrace();
        

        return "ok";