多线程保存数据到数据库

  List<PostAddress>  postAddressDetails = postAddressDao.getAllList();
          if(CollectionUtils.isEmpty(postAddressDetails)){
              return ;
          }
        Set<Long> list =new HashSet<>();
        for (PostAddress postAddress : postAddressDetails) {
            list.add(postAddress.getCustomer_uid());
        }
        for (Long uid : list) {
            addressCacheService.delete(uid);
        }

        if (postAddressDetails.size() > 10000) {

            betch = 10;

        } else if (postAddressDetails.size() > 5000) {

            betch = 5;

        } else if (postAddressDetails.size() > 1000) {
            betch = 3;
        }
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(betch);
        float num = (float)  postAddressDetails.size()/500;
        double ceil = Math.ceil(num);
        for (int i = 0; i < ceil; i++) {
            int startLen = i * 500;
            int endLen = ((i + 1) * 500 > postAddressDetails.size() ? postAddressDetails.size() - 1 : (i + 1) * 500);
            // 该线程处理
            final List<PostAddress> postAddresses = postAddressDetails.subList(startLen, endLen);
            fixedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    for (PostAddress postAddressDetail : postAddresses) {
                        add(postAddressDetail);
                    }
                }
            });

        }

展开全部

以mysql为数据库32313133353236313431303231363533e4b893e5b19e31333332616431写的一个粗陋的demo，你参考一下，希望不会因为代码过多被百度吞了——

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.util.ArrayList;

import java.util.List;

public class Test {

public static void main(String[] args) {

allotThread();

}

/**

* 将100条数据分成10份并启动10个线程分别操作

public static void allotThread() {

List datas = buildDatas();

for (int i=0; i<100; i+=10) {

List tenDatas = datas.subList(i, i + 10);

insertData(tenDatas);

}

/**

* 创建100条模拟数据

* @return

public static List buildDatas() {

List datas = new ArrayList();

for (int i=0; i<100; i++) {

String[] data = {"id " + i, "name " + i};

datas.add(data);

}

return datas;

}

/**

* 启动线程进行数据插入操作

* @param tenDatas

public static void insertData(final List tenDatas) {

new Thread(new Runnable() {

public void run() {

String sql = "insert into testtable (id, name) values (?, ?)";

Connection conn = null;

PreparedStatement pstmt = null;

try {

conn = getConnection();

conn.setAutoCommit(false);

pstmt = getPstmt(conn, sql);

for (String[] data : tenDatas) {

pstmt.setString(1, data[0]);

pstmt.setString(2, data[1]);

pstmt.addBatch();

}

pstmt.executeBatch();

conn.commit();

conn.setAutoCommit(true);

} catch (SQLException e) {

e.printStackTrace();

rollback(conn);

} catch (ClassNotFoundException e) {

e.printStackTrace();

} finally {

close(pstmt);

close(conn);

}

}).start();

}

public static Connection getConnection() throws SQLException, ClassNotFoundException {

Class.forName("com.mysql.jdbc.Driver");

String dbUrl = "jdbc:mysql://localhost/test?useUnicode=true&characterEncoding=UTF-8";

Connection conn = DriverManager.getConnection(dbUrl, "root", "tooeasy");

return conn;

}

public static PreparedStatement getPstmt(Connection conn, String sql) throws SQLException, ClassNotFoundException {

PreparedStatement pstmt = conn.prepareStatement(sql);

return pstmt;

}

public static void rollback(Connection conn) {

try {

if (null != conn) {

conn.rollback();

}

} catch (SQLException e) {

e.printStackTrace();

}

public static void close(Connection conn) {

try {

if (null != conn) {

conn.close();

}

} catch (SQLException e) {

e.printStackTrace();

}

public static void close(PreparedStatement pstmt) {

try {

if (null != pstmt) {

pstmt.close();

}

} catch (SQLException e) {

e.printStackTrace();

}

public static void close(ResultSet rs) {

try {

if (null != rs) {

rs.close();

}

} catch (SQLException e) {

e.printStackTrace();

}

萌新在写网站的发送邮件验证，为了防止用户滥发，所以加了权限。前端简单地disable按钮一刷新就没了，纯粹视觉提示作用，所以在后端models里为user加了一个resend_right，当为True时才能重新发送，False不行。

所以在models里，user模型有一个column是这样的(SQLAlchemy)：

resend_right = db.Column(db.Boolean, default=True)

当然前端是等待60秒后可以重新发送，所以后端也计时60秒后重新赋值True给resend_right。我就想这种等待性IO/数据库读取录入等操作当然是多线程处理。

所以我写了resend_right权限重置的方法：

def async_reset(app, user):

with app.app_context():

time.sleep(55)

user.resend_right = True

def resend_right_reset(user):

app = current_app._get_current_object()

thr = Thread(target=async_reset, args=[app, user])

thr.start()

return thr

然后在views的路由函数里面调用它：

# Resend confirmation email route, need to be protected

@auth.route('/resend_email/')

@login_required

def resend_confirmation():

mail_host ='http://mail.' + re.split('@', current_user.email)[1]

if not current_user.resend_right:

flash("请不要尝试刷新页面来短时间内重复发送验证邮件，你可以在一分钟后再试")

return render_template('auth/confirm.html',user=current_user, mail_host=mail_host)

token = current_user.generate_confirmation_token()

.........

结果无效，所以我测试了一下，发现路由函数无问题,resend_right_reset无问题。假如我把user.rend_right=True写进resend_right_reset是能够正常运作的，但一旦用多线程来处理就始终无法重置。然后我分析，多线程这里用了current_app._get_current_object()获取全局对象，然后app.app_context()拿到了上下文导入到多线程里，应该就没问题了。但为什么不行？

求教，非常感谢！


 任务: 从sqlserver中将一个表A(约16W条数据)导到mysql中对应的一个表B中。
 思路：分段获取A表中的数据后，用多个线程同时向B表中写入。
 关键代码
 //将数据库中的数据条数分段
 public void division(){
 //获取要导入的总的数据条数
 String sql3="SELECT count(*) FROM [CMD].[dbo].[mycopy1]";
 try {
 pss=cons.prepareStatement(sql3);
 rss=pss.executeQuery();
 while(rss.next()){
 System.out.println("总记录条数:"+rss.getInt(1));
 sum=rss.getInt(1);
 }
 //每30000条记录作为一个分割点
 if(sum>=30000){
 n=sum/30000;
 residue=sum%30000;
 }else{
 residue=sum;
 }
 System.out.println(n+" "+residue);
 } catch (SQLException e) {
 // TODO Auto-generated catch block
 e.printStackTrace();
 }
 } 线程类
 public MyThread(int start,int end) {
 this.end=end;
 this.start=start;
 System.out.println("处理掉余数");
 try {
 System.out.println("--------"+Thread.currentThread().getName()+"------------");
 Class.forName(SQLSERVERDRIVER);
 System.out.println("加载sqlserver驱动...");
 cons = DriverManager.getConnection(CONTENTS,UNS,UPS);
 stas = cons.createStatement();
 System.out.println("连接SQLServer数据库成功！！");
 System.out.println("加载mysql驱动.....");
 Class.forName(MYSQLDRIVER);
 con = DriverManager.getConnection(CONTENT,UN,UP);
 sta = con.createStatement();
 // 关闭事务自动提交
 con.setAutoCommit(false);
 System.out.println("连接mysql数据库成功！！");
 } catch (Exception e) {
 e.printStackTrace();
 }
 // TODO Auto-generated constructor stub
 }
 public ArrayList getAll(){
 Member member;
 String sql1="select * from (select row_number() over (order by pmcode) as rowNum,*" +
 " from [CMD].[dbo].[mycopy1]) as t where rowNum between "+start+" and "+end;
 try {
 System.out.println("正在获取数据...");
 allmembers=new ArrayList();
 rss=stas.executeQuery(sql1);
 while(rss.next()){
 member=new Member();
 member.setAddress1(rss.getString("address1"));
 member.setBnpoints(rss.getString("bnpoints"));
 member.setDbno(rss.getString("dbno"));
 member.setExpiry(rss.getString("expiry"));
 member.setHispoints(rss.getString("hispoints"));
 member.setKypoints(rss.getString("kypoints"));
 member.setLevels(rss.getString("levels"));
 member.setNames(rss.getString("names"));
 member.setPmcode(rss.getString("pmcode"));
 member.setRemark(rss.getString("remark"));
 member.setSex(rss.getString("sex"));
 member.setTelephone(rss.getString("telephone"));
 member.setWxno(rss.getString("wxno"));
 member.setPmdate(rss.getString("pmdate"));
 allmembers.add(member);
 //System.out.println(member.getNames());
 }
 System.out.println("成功获取sqlserver数据库数据！");
 return allmembers;
 } catch (SQLException e) {
 // TODO Auto-generated catch block
 System.out.println("获取sqlserver数据库数据发送异常！");
 e.printStackTrace();
 }
 try {
 rss.close();
 stas.close();
 } catch (SQLException e) {
 // TODO Auto-generated catch block
 e.printStackTrace();
 }
 return null;
 }
 public void inputAll(ArrayList allmembers){
 System.out.println("开始向mysql中写入");
 String sql2="insert into test.mycopy2 values (?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
 try {
 ps=con.prepareStatement(sql2);
 System.out.println("-------------------------等待写入数据条数: "+allmembers.size());
 for(int i=0;i
 ps.setString(1, allmembers.get(i).getPmcode());
 ps.setString(2, allmembers.get(i).getNames());
 //System.out.println(allmembers.get(i).getNames());
 ps.setString(3, allmembers.get(i).getSex());
 ps.setString(4, allmembers.get(i).getTelephone());
 ps.setString(5, allmembers.get(i).getAddress1());
 ps.setString(6, allmembers.get(i).getPmdate());
 ps.setString(7, allmembers.get(i).getExpiry());
 ps.setString(8, allmembers.get(i).getLevels());
 ps.setString(9, allmembers.get(i).getDbno());
 ps.setString(10, allmembers.get(i).getHispoints());
 ps.setString(11, allmembers.get(i).getBnpoints());
 ps.setString(12, allmembers.get(i).getKypoints());
 ps.setString(13, allmembers.get(i).getWxno());
 ps.setString(14, allmembers.get(i).getRemark());
 //插入命令列表
 //ps.addBatch();
 ps.executeUpdate();
 }
 //ps.executeBatch();
 con.commit();
 ps.close();
 con.close();
 this.flag=false;
 System.out.println(Thread.currentThread().getName()+"--->OK");
 } catch (SQLException e) {
 // TODO Auto-generated catch block
 System.out.println("向mysql中更新数据时发生异常！");
 e.printStackTrace();
 }
 }
 @Override
 public void run() {
 // TODO Auto-generated method stub
 while(true&&flag){
 this.inputAll(getAll());
 }
 }
 测试类：
 public class Test1 {
 DbManager dm=null;
 MyThread my1=null;
 public Test1(){
 dm=new DbManager();
 System.out.println(dm.n+"----"+dm.residue);
 if(dm.n<1){//数据条数小于30000单线程处理
 my1=new MyThread(1,dm.sum);
 my1.start=1;
 my1.end=dm.residue;
 Thread t1=new Thread(my1);
 t1.start();
 }else{//大于30000时
 //起n个线程 每个处理30000条数据
 for (int i = 1; i <=dm.n; i++) {
 new Thread(new MyThread(i)).start();
 try {
 Thread.sleep(1);
 } catch (InterruptedException e) {
 // TODO Auto-generated catch block
 e.printStackTrace();
 }
 }
 //处理掉余数
 my1=new MyThread(dm.n*30000+1,dm.sum);
 Thread t1=new Thread(my1);
 t1.start();
 }
 }
 public static void main(String[] args) {
 //new Test1();
 //迁移完数据，自动关机
 try {
 Runtime.getRuntime().exec("cmd /c Shutdown -t 10");
 } catch (IOException e) {
 // TODO Auto-generated catch block
 e.printStackTrace();
 }
 }
 经过多次测试：从sqlserver中读取16w条数据并写入mysql，耗时15min左右。
 开始会报错： java heap space
 解决方案：(myeclipse)window->Preferences->Java->Installed JREs，选择当前的JRE，然后edit它；在新窗口里设置Default VM Arguments为 -Xms512M -Xmx512M即可 ：

收起

opencv3和opencv4多线程内存泄漏问题

Java 面经手册·小傅哥.pdf

CNN人脸识别签到系统源文件+报告论文

基于C++/Python的量化交易研究框架.rar

多线程操作数据写入数据库

java 多线程存储数据库

在flask框架下利用Python的threading或thread多线程库如何操作数据库？

java多线程向数据库写入数据

Python基于多线程实现抓取数据存入数据库的方法

java多线程并行处理数据丢失，写给正在求职的Java开发 后端 java

Java多线程向数据库写入数据

java通过poi上传excel表格并多线程批量插入数据到数据库

JPA 多线程对数据进行更新，Update和Insert同时存在的问题

在多线程下不能操作数据库么？

多线程批量插入数据库 java

Springboot多线程异步批量添加/导入大数据到数据库 java 多线程 并发编程

java – 多线程 – 避免和处理数据库死锁

多线程执行List导入Mysql数据库方法 数据库 list mysql

易语言多线程查询数据库易语言多线程并发

Android多线程操作sqlite数据库连接池框架的一种设计思路

干货~多线程下1分钟完成1000万条数据插入到数据库中 java 多线程

如何保证多线程从mysql数据库查询的数据不重复

请教多线程数据库程序如何保证线程安全

多线程实现数据库的并发操作

Java 多线程事务回滚 ——多线程插入数据库时事务控制（）

多线程批量拆分List导入数据库 python java 数据库 sqlserver 人工智能

多线程批量插入数据到mysql java java-ee 开发语言

springboot多线程读取百万级数据写入文件的代码实例

一个txt文件，里面存在几万条数据，现用多线程实现数据导入，提高效率，java代码如何实现

多线程实现大批量数据导入

用多线程写入数据库的问题(150分) 线程池写入mysql

Android：在SQLite数据库上编写多个线程

ssm框架中使用多线程批量添加数据

多线程保存数据到数据库

opencv3和opencv4多线程内存泄漏问题

Java 面经手册·小傅哥.pdf

CNN人脸识别签到系统源文件+报告论文

基于C++/Python的量化交易研究框架.rar

多线程操作数据写入数据库

java 多线程存储数据库

在flask框架下利用Python的threading或thread多线程库如何操作数据库？

java多线程向数据库写入数据

Python基于多线程实现抓取数据存入数据库的方法

Java多线程向数据库写入数据

java通过poi上传excel表格并多线程批量插入数据到数据库

JPA 多线程对数据进行更新，Update和Insert同时存在的问题

在多线程下不能操作数据库么？

java – 多线程 – 避免和处理数据库死锁

易语言多线程查询数据库 易语言多线程并发

Android多线程操作sqlite数据库连接池框架的一种设计思路

如何保证多线程从mysql数据库查询的数据不重复

请教多线程数据库程序如何保证线程安全

多线程实现数据库的并发操作

Java 多线程事务回滚 ——多线程插入数据库时事务控制（）

springboot多线程读取百万级数据写入文件的代码实例

一个txt文件，里面存在几万条数据，现用多线程实现数据导入，提高效率，java代码如何实现

多线程实现大批量数据导入

Android：在SQLite数据库上编写多个线程

ssm框架中使用多线程批量添加数据

多线程保存数据到数据库

易语言多线程查询数据库易语言多线程并发