1.如何分批处理数据？
1.使用LIMIT和OFFSET子句： 这是最常用的分批查询方法。例如，你可以使用以下SQL语句来分批查询数据：
SELECT * FROM your_table LIMIT 1000 OFFSET 0;分批查询到的数据在后端进行处理，达到分批处理数据的效果。
2.使用多线程的方式： 如果你需要用多线程分批处理数据，并且数据所在表的主键id是递增的，可以使用取模的方式进行分批查询。例如：
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
public class DatabaseUtils {
// 数据库连接信息
private static final String URL = "jdbc:mysql://localhost:3306/your_database";
private static final String USER = "your_username";
private static final String PASSWORD = "your_password";
// 获取数据库连接
public static Connection getConnection() throws SQLException {
return DriverManager.getConnection(URL, USER, PASSWORD);
}
// 异步查询数据库的方法
//第一个参数表示偏移量，表示当前已经查询到的数据id
//第二个参数表示从当前偏移量开始，查询多少条数据
public static CompletableFuture<List<String>> queryBatchAsync(int offset, int limit) {
// 使用CompletableFuture.supplyAsync来异步执行数据库查询
return CompletableFuture.supplyAsync(() -> {
List<String> results = new ArrayList<>();
try (Connection conn = getConnection();
PreparedStatement stmt = conn.prepareStatement("SELECT id, data FROM your_table LIMIT ? OFFSET ?")) {
// 设置查询的LIMIT和OFFSET
stmt.setInt(1, limit);
stmt.setInt(2, offset);
// 执行查询
try (ResultSet rs = stmt.executeQuery()) {
// 遍历结果集，将结果添加到列表中
while (rs.next()) {
results.add(rs.getString("id") + ": " + rs.getString("data"));
}
}
} catch (SQLException e) {
// 如果发生异常，抛出运行时异常
throw new RuntimeException(e);
}
// 返回查询结果
return results;
});
}
}这个类只是负责连接数据库，以及一个异步查询数据库的方法。注意这个方法的返回结果是CompletableFuture<List<String>>，返回一个异步任务，异步任务中的返回结果是根据偏移量和批量查询条数的查询结果，封装成一个list集合。注意数据库中的id应该是自增的
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class MultiThreadedBatchProcessing {
public static void main(String[] args) {
// 假设我们有1000条记录需要处理，每批处理100条记录
int totalRecords = 1000;
int batchSize = 100;
// 创建一个有10个线程的线程池
ExecutorService executor = Executors.newFixedThreadPool(10);
// 创建一个CompletableFuture数组来存储每个批次的异步任务
CompletableFuture<?>[] futures = new CompletableFuture[10];
// 循环创建并启动每个批次的异步查询任务
for (int i = 0; i < totalRecords; i += batchSize) {
int offset = i; // 计算当前批次的起始位置
int limit = batchSize; // 每批处理的记录数
// 启动异步查询任务
futures[i / batchSize] = DatabaseUtils.queryBatchAsync(offset, limit).thenAccept(batchResult -> {
// 处理每个批次的结果
for (String record : batchResult) {
System.out.println(record);
}
});
}
// 使用CompletableFuture.allOf等待所有批次的任务完成
CompletableFuture.allOf(futures).thenRun(() -> {
// 所有批次处理完成后，关闭线程池
System.out.println("All batches have been processed.");
executor.shutdown();
}).exceptionally(e -> {
// 如果发生异常，打印错误信息，并尝试紧急关闭线程池
System.err.println("An error occurred: " + e.getMessage());
executor.shutdownNow();
return null;
});
}
}追问：若多线程分批查询过程中有数据插入或者删除，则数据缺漏，如何解决问题？
使用事务保证数据一致性： 可以通过事务来确保数据的一致性。在事务中执行查询、插入或删除操作，如果中途发生错误，可以通过回滚操作来撤销所有已执行的步骤，确保数据的完整性。这样可以避免因并发操作导致的数据不一致问题。
追问：多线程共享事务存在问题，不合适，有其他方式吗？
- 消息队列和异步重试：在执行更新数据库和删除缓存的操作时，可以使用消息队列和异步重试机制。这样，即使某个操作失败，也可以通过消息队列进行补偿操作，确保数据的最终一致性。
分布式锁：在高并发场景下，可以使用分布式锁来保证同一时间只有一个线程能修改特定的数据行。这可以通过在应用程序层面采用分布式锁、Redis等中间件实现锁机制来完成
