devbox
算法编织者2
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

关于jdbc批量操作(addBatch, executeBatch)的测试

作者：算法编织者2 | 2024-02-05 22:18:57

赞

踩

executebatch

由于项目上有大批量数据插入和更新的操作，所以使用了jdbc的批量操作功能。在此之前参考了很多文章包括jdbc的手册(https://www.tutorialspoint.com/jdbc/jdbc-batch-processing.htm)，有说需要使用事务的，也有的文章没有使用事务，试了很久，发现代码一直没有按照期望的执行，还是逐条操作数据库。后来在偶然间看到了文章提到需要设置rewriteBatchedStatements为true，加上该参数后搞定！这里记录一下我所做个几个场景的实验，并配有抓包的记录，以显示jdbc批量操作的过程。


测试使用的表结构如下：

  1. create table employees (
  2.   id int(11) unsigned not null auto_increment,
  3.   user_id int(20) not null,
  4.   age int(10) not null,
  5.   first_name varchar(20) not null,
  6.   second_name varchar(20) not null,
  7.   date date not null,
  8.   PRIMARY KEY (id)
  9. ) ENGINE=InnoDB CHARSET=utf8;

使用tcpdump抓包，并在wireshark下做分析


场景一：不使用事务，不添加rewriteBatchedStatements=true参数

代码隐去数据库ip、库名、账户名和密码

  1. package jdbcbatchtest;
  2.  
  3.  
  4. import java.sql.Connection;
  5. import java.sql.DriverManager;
  6. import java.sql.PreparedStatement;
  7. import java.sql.SQLException;
  8.  
  9. public class Main {
  10. 	public static void main(String[] args) {
  11. 		Connection conn = null;
  12. 		PreparedStatement pst = null;
  13. 		
  14. 		try {
  15. 			Class.forName("com.mysql.jdbc.Driver");
  16. 			conn = DriverManager.getConnection("jdbc:mysql://********:3306/****", "****", "****");
  17. 			
  18. 			String sql = "insert into employees (user_id, age, first_name, second_name, date) values(?,?,?,?,?)";
  19. 			
  20. 			pst = conn.prepareStatement(sql);
  21. 			
  22. 			int loop = 0;
  23. 			for(loop = 0; loop < 1000; loop++) {
  24. 				pst.setInt(1, loop);
  25. 				pst.setInt(2, 18);
  26. 				pst.setString(3, "roger");
  27. 				pst.setString(4, "zhang");
  28. 				pst.setString(5, "2017-01-17");
  29. 				pst.addBatch();
  30. 			}
  31.  
  32. 			pst.executeBatch();
  33. 		} catch (ClassNotFoundException e) {
  34. 			e.printStackTrace();
  35. 		} catch (SQLException e) {
  36. 			e.printStackTrace();
  37. 		} finally {
  38. 			if(pst != null) {
  39. 				try {
  40. 					pst.close();
  41. 				} catch (SQLException e) {
  42. 					e.printStackTrace();
  43. 				}
  44. 			}
  45. 			
  46. 			if(conn != null) {
  47. 				try {
  48. 					conn.close();
  49. 				} catch (SQLException e) {
  50. 					e.printStackTrace();
  51. 				}
  52. 			}
  53. 		}
  54. 	}
  55. }
  56.

抓包结果:


从图片中的抓包结果可以看出，sql语句是逐条被提交到mysql服务器的，该操作一共执行了1000次。

场景二：使用事务，不添加rewriteBatchedStatements=true参数

  1. package jdbcbatchtest;
  2.  
  3. import java.sql.Connection;
  4. import java.sql.DriverManager;
  5. import java.sql.PreparedStatement;
  6. import java.sql.SQLException;
  7.  
  8.  
  9. public class Main {
  10.  
  11. 	public static void main(String[] args) {
  12. 		Connection conn = null;
  13. 		PreparedStatement pst = null;
  14. 		
  15. 		try {
  16. 			Class.forName("com.mysql.jdbc.Driver");
  17. 			conn = DriverManager.getConnection("jdbc:mysql://********:3306/****", "****", "****");
  18. 			
  19. 			String sql = "insert into employees (user_id, age, first_name, second_name, date) values(?,?,?,?,?)";
  20. 			
  21. 			conn.setAutoCommit(false);
  22. 			pst = conn.prepareStatement(sql);
  23. 			
  24. 			int loop = 0;
  25. 			for(loop = 0; loop < 1000; loop++) {
  26. 				pst.setInt(1, loop);
  27. 				pst.setInt(2, 18);
  28. 				pst.setString(3, "roger");
  29. 				pst.setString(4, "zhang");
  30. 				pst.setString(5, "2017-01-17");
  31. 				pst.addBatch();
  32. 			}
  33.  
  34. 			pst.executeBatch();
  35. 			conn.commit();
  36. 		} catch (ClassNotFoundException e) {
  37. 			e.printStackTrace();
  38. 		} catch (SQLException e) {
  39. 			e.printStackTrace();
  40. 			try {
  41. 				conn.rollback();
  42. 			} catch (SQLException e1) {
  43. 				e1.printStackTrace();
  44. 			}
  45. 		} finally {
  46. 			if(pst != null) {
  47. 				try {
  48. 					pst.close();
  49. 				} catch (SQLException e) {
  50. 					e.printStackTrace();
  51. 				}
  52. 			}
  53. 			
  54. 			if(conn != null) {
  55. 				try {
  56. 					conn.close();
  57. 				} catch (SQLException e) {
  58. 					e.printStackTrace();
  59. 				}
  60. 			}
  61. 		}
  62. 	}
  63. }


抓包结果：


和场景一一样，sql语句还是逐条发送到mysql服务器，不同点在于最有有一条commit的数据包，提交事务。

场景三：不使用事务，添加rewriteBatchedStatements=true参数

  1. package jdbcbatchtest;
  2.  
  3. import java.sql.Connection;
  4. import java.sql.DriverManager;
  5. import java.sql.PreparedStatement;
  6. import java.sql.SQLException;
  7.  
  8. public class Main {
  9.  
  10. 	public static void main(String[] args) {
  11. 		Connection conn = null;
  12. 		PreparedStatement pst = null;
  13. 		
  14. 		try {
  15. 			Class.forName("com.mysql.jdbc.Driver");
  16. 			conn = DriverManager.getConnection("jdbc:mysql://********:3306/****?rewriteBatchedStatements=true", "****", "****");
  17. 			
  18. 			String sql = "insert into employees (user_id, age, first_name, second_name, date) values(?,?,?,?,?)";
  19. 			
  20. 			pst = conn.prepareStatement(sql);
  21. 			
  22. 			int loop = 0;
  23. 			for(loop = 0; loop < 1000; loop++) {
  24. 				pst.setInt(1, loop);
  25. 				pst.setInt(2, 18);
  26. 				pst.setString(3, "roger");
  27. 				pst.setString(4, "zhang");
  28. 				pst.setString(5, "2017-01-17");
  29. 				pst.addBatch();
  30. 			}
  31.  
  32. 			pst.executeBatch();
  33. 		} catch (ClassNotFoundException e) {
  34. 			e.printStackTrace();
  35. 		} catch (SQLException e) {
  36. 			e.printStackTrace();
  37. 		} finally {
  38. 			if(pst != null) {
  39. 				try {
  40. 					pst.close();
  41. 				} catch (SQLException e) {
  42. 					e.printStackTrace();
  43. 				}
  44. 			}
  45. 			
  46. 			if(conn != null) {
  47. 				try {
  48. 					conn.close();
  49. 				} catch (SQLException e) {
  50. 					e.printStackTrace();
  51. 				}
  52. 			}
  53. 		}
  54. 	}
  55. }


抓包结果：


从抓包结果可以看出，jdbc将1000条insert语句拆分成了10条报文分批发送到mysql服务器（这里做了几次试验发现每次操作报文的大小和个数都不是固定的），每发送一次报文便插入一批数据进入数据库，实现了批量的操作。这里需要注意的是，在我的理解看来，这10条消息是立即生效的，也就是说如果中间某条消息中的插入操作发生了异常，那么之前的操作是无法回滚的。这也便引出了下面的第四种场景。

场景四：使用事务，添加rewriteBatchedStatements=true参数

  1. package jdbcbatchtest;
  2.  
  3. import java.sql.Connection;
  4. import java.sql.DriverManager;
  5. import java.sql.PreparedStatement;
  6. import java.sql.SQLException;
  7.  
  8. public class Main {
  9.  
  10. 	public static void main(String[] args) {
  11. 		Connection conn = null;
  12. 		PreparedStatement pst = null;
  13. 		
  14. 		try {
  15. 			Class.forName("com.mysql.jdbc.Driver");
  16. 			conn = DriverManager.getConnection("jdbc:mysql://********:3306/****?rewriteBatchedStatements=true", "****", "****");
  17. 			
  18. 			String sql = "insert into employees (user_id, age, first_name, second_name, date) values(?,?,?,?,?)";
  19. 			
  20. 			conn.setAutoCommit(false);
  21. 			pst = conn.prepareStatement(sql);
  22. 			
  23. 			int loop = 0;
  24. 			for(loop = 0; loop < 1000; loop++) {
  25. 				pst.setInt(1, loop);
  26. 				pst.setInt(2, 18);
  27. 				pst.setString(3, "roger");
  28. 				pst.setString(4, "zhang");
  29. 				pst.setString(5, "2017-01-17");
  30. 				pst.addBatch();
  31. 			}
  32.  
  33. 			pst.executeBatch();
  34. 			conn.commit();
  35. 		} catch (ClassNotFoundException e) {
  36. 			e.printStackTrace();
  37. 		} catch (SQLException e) {
  38. 			e.printStackTrace();
  39. 			try {
  40. 				conn.rollback();
  41. 			} catch (SQLException e1) {
  42. 				e1.printStackTrace();
  43. 			}
  44. 		} finally {
  45. 			if(pst != null) {
  46. 				try {
  47. 					pst.close();
  48. 				} catch (SQLException e) {
  49. 					e.printStackTrace();
  50. 				}
  51. 			}
  52. 			
  53. 			if(conn != null) {
  54. 				try {
  55. 					conn.close();
  56. 				} catch (SQLException e) {
  57. 					e.printStackTrace();
  58. 				}
  59. 			}
  60. 		}
  61. 	}
  62. }

抓包结果：


和场景三相比，也是将1000条sql语句分成若干个报文发送到mysql服务器，只是最后多了一个commit的操作。


综上，rewriteBatchedStatements=true才是jdbc实现批量操作的关键。


通过对源码的分析，我们来更加深入地理解一下其中的原理。(源码来自于github, 版本为5.1)

我们可以在StatementImpl.java中找到答案，在executeBatchInternal这个函数中有这么一段：

  1.     if (this.batchedArgs != null) {
  2.         int nbrCommands = this.batchedArgs.size();
  3.  
  4.         this.batchedGeneratedKeys = new ArrayList<ResultSetRow>(this.batchedArgs.size());
  5.  
  6.         boolean multiQueriesEnabled = locallyScopedConn.getAllowMultiQueries();
  7.  
  8.         if (locallyScopedConn.versionMeetsMinimum(4, 1, 1)
  9.              && (multiQueriesEnabled || (locallyScopedConn.getRewriteBatchedStatements() && nbrCommands > 4))) {
  10.             return executeBatchUsingMultiQueries(multiQueriesEnabled, nbrCommands, individualStatementTimeout);
  11.         }
  12.  
  13.         if (locallyScopedConn.getEnableQueryTimeouts() && individualStatementTimeout != 0 && locallyScopedConn.versionMeetsMinimum(5, 0, 0)) {
  14.             timeoutTask = new CancelTask(this);
  15.             locallyScopedConn.getCancelTimer().schedule(timeoutTask, individualStatementTimeout);
  16.         }


  1.     public boolean getRewriteBatchedStatements() {
  2.         return this.rewriteBatchedStatements.getValueAsBoolean();
  3.     }


由于multiQueriesEnabled的默认值是false, 通过locallyScopedConn.getRewriteBatchedStatements()我们得知 只有当rewriteBatchedStatements标志位为true并且一次的批量 操作的数据大于4时，才会使用批量操作。 

  1.     private BooleanConnectionProperty rewriteBatchedStatements = new BooleanConnectionProperty("rewriteBatchedStatements", false,
  2.             Messages.getString("ConnectionProperties.rewriteBatchedStatements"), "3.1.13", PERFORMANCE_CATEGORY, Integer.MIN_VALUE);

而rewriteBatchedStatements的默认值为false，这就是本文之前提到的在批量操作时为什么要设置该参数为true的原因。


我们再来看看jdbc具体执行批量操作的代码executeBatchUsingMultiQueries:

  1.     for (commandIndex = 0; commandIndex < nbrCommands; commandIndex++) {
  2.         String nextQuery = (String) this.batchedArgs.get(commandIndex);
  3.  
  4.         if (((((queryBuf.length() + nextQuery.length()) * numberOfBytesPerChar) + 1 /* for semicolon */
  5.             + MysqlIO.HEADER_LENGTH) * escapeAdjust) + 32 > this.connection.getMaxAllowedPacket()) {
  6.             try {
  7.                 batchStmt.execute(queryBuf.toString(), java.sql.Statement.RETURN_GENERATED_KEYS);
  8.             } catch (SQLException ex) {
  9.                 sqlEx = handleExceptionForBatch(commandIndex, argumentSetsInBatchSoFar, updateCounts, ex);
  10.             }
  11.  
  12.             counter = processMultiCountsAndKeys((StatementImpl) batchStmt, counter, updateCounts);
  13.  
  14.             queryBuf = new StringBuilder();
  15.             argumentSetsInBatchSoFar = 0;
  16.         }
  17.  
  18.         queryBuf.append(nextQuery);
  19.         queryBuf.append(";");
  20.         argumentSetsInBatchSoFar++;
  21.     }

当一个数据包的长度不超过maxAllowedPacket，会持续累加，直到超过最大长度时将数据包发送出去。


拓展: multiQueriesEnabled的作用

multiQueriesEnabled为true，则jdbc支持执行的一条语句中包含多条由分号分割的语句。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/算法编织者2/article/detail/61872
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号