Browse Source

mysql存s3初步完成

master
review512jwy@163.com 2 weeks ago
parent
commit
7d19facc3b
  1. 70
      src/main/java/com/dashboard/aws/lambda/handler/MySQLBatchToS3Handler.java
  2. 2
      src/main/java/com/dashboard/aws/lambda/service/MySQLService.java

70
src/main/java/com/dashboard/aws/lambda/handler/MySQLBatchToS3Handler.java

@ -42,15 +42,39 @@ public class MySQLBatchToS3Handler implements RequestHandler<Map<String, Object>
* 失败记录数据结构 * 失败记录数据结构
*/ */
static class FailedRecord { static class FailedRecord {
public List<Long> ids; private List<Long> ids;
public int retryCount; private int retryCount;
public List<Long> getIds() {
return ids;
}
public void setIds(List<Long> ids) {
this.ids = ids;
}
public int getRetryCount() {
return retryCount;
}
public void setRetryCount(int retryCount) {
this.retryCount = retryCount;
}
} }
/** /**
* 失败记录集合 * 失败记录集合
*/ */
static class FailedRecords { static class FailedRecords {
public Map<String, FailedRecord> tableRecords = new HashMap<>(); private Map<String, FailedRecord> tableRecords = new HashMap<>();
public Map<String, FailedRecord> getTableRecords() {
return tableRecords;
}
public void setTableRecords(Map<String, FailedRecord> tableRecords) {
this.tableRecords = tableRecords;
}
} }
@Override @Override
@ -73,7 +97,7 @@ public class MySQLBatchToS3Handler implements RequestHandler<Map<String, Object>
// 加载失败记录 // 加载失败记录
FailedRecords failedRecords = loadFailedRecords(); FailedRecords failedRecords = loadFailedRecords();
logger.info("Loaded failed records for {} tables", failedRecords.tableRecords.size()); logger.info("Loaded failed records for {} tables", failedRecords.getTableRecords().size());
// 处理每个配置的表 // 处理每个配置的表
for (String table : Constants.tables) { for (String table : Constants.tables) {
@ -223,14 +247,14 @@ public class MySQLBatchToS3Handler implements RequestHandler<Map<String, Object>
if (!failedIds.isEmpty()) { if (!failedIds.isEmpty()) {
logger.error("[{}][{}] delete failed for {} records in date {}, recording to failed records", logger.error("[{}][{}] delete failed for {} records in date {}, recording to failed records",
DB_SCHEMA, table, failedIds.size(), dateKey); DB_SCHEMA, table, failedIds.size(), dateKey);
FailedRecord record = failedRecords.tableRecords.get(table); FailedRecord record = failedRecords.getTableRecords().get(table);
if (record == null) { if (record == null) {
record = new FailedRecord(); record = new FailedRecord();
record.ids = new ArrayList<>(); record.setIds(new ArrayList<>());
record.retryCount = 0; record.setRetryCount(0);
failedRecords.tableRecords.put(table, record); failedRecords.getTableRecords().put(table, record);
} }
record.ids.addAll(failedIds); record.getIds().addAll(failedIds);
anyFailure = true; anyFailure = true;
} }
} catch (Exception e) { } catch (Exception e) {
@ -315,7 +339,7 @@ public class MySQLBatchToS3Handler implements RequestHandler<Map<String, Object>
try { try {
byte[] content = objectMapper.writeValueAsBytes(failedRecords); byte[] content = objectMapper.writeValueAsBytes(failedRecords);
s3Service.upload(s3Key, content); s3Service.upload(s3Key, content);
logger.info("Saved failed records, total tables: {}", failedRecords.tableRecords.size()); logger.info("Saved failed records, total tables: {}", failedRecords.getTableRecords().size());
} catch (Exception e) { } catch (Exception e) {
logger.error("Failed to save failed records", e); logger.error("Failed to save failed records", e);
} }
@ -325,43 +349,43 @@ public class MySQLBatchToS3Handler implements RequestHandler<Map<String, Object>
* 处理失败记录重试删除 * 处理失败记录重试删除
*/ */
private void processFailedRecords(FailedRecords failedRecords, String table) { private void processFailedRecords(FailedRecords failedRecords, String table) {
FailedRecord record = failedRecords.tableRecords.get(table); FailedRecord record = failedRecords.getTableRecords().get(table);
if (record == null || record.ids == null || record.ids.isEmpty()) { if (record == null || record.getIds() == null || record.getIds().isEmpty()) {
logger.info("[{}][{}] No failed records to process", DB_SCHEMA, table); logger.info("[{}][{}] No failed records to process", DB_SCHEMA, table);
return; return;
} }
logger.info("[{}][{}] Processing {} failed records, retry count: {}", logger.info("[{}][{}] Processing {} failed records, retry count: {}",
DB_SCHEMA, table, record.ids.size(), record.retryCount); DB_SCHEMA, table, record.getIds().size(), record.getRetryCount());
int maxRetries = 10; int maxRetries = 10;
if (record.retryCount >= maxRetries) { if (record.getRetryCount() >= maxRetries) {
logger.info("[{}][{}] Failed records reached max retries ({}), removing", logger.info("[{}][{}] Failed records reached max retries ({}), removing",
DB_SCHEMA, table, maxRetries); DB_SCHEMA, table, maxRetries);
failedRecords.tableRecords.remove(table); failedRecords.getTableRecords().remove(table);
return; return;
} }
// 尝试删除 // 尝试删除
List<Long> stillFailedIds = mysqlService.deleteByIds(DB_SCHEMA, table, record.ids); List<Long> stillFailedIds = mysqlService.deleteByIds(DB_SCHEMA, table, record.getIds());
if (stillFailedIds.isEmpty()) { if (stillFailedIds.isEmpty()) {
// 全部成功 // 全部成功
logger.info("[{}][{}] Success delete {} failed records, removing from list", logger.info("[{}][{}] Success delete {} failed records, removing from list",
DB_SCHEMA, table, record.ids.size()); DB_SCHEMA, table, record.getIds().size());
failedRecords.tableRecords.remove(table); failedRecords.getTableRecords().remove(table);
} else { } else {
// 部分失败 // 部分失败
record.retryCount++; record.setRetryCount(record.getRetryCount() + 1);
record.ids = stillFailedIds; record.setIds(stillFailedIds);
logger.error("[{}][{}] Still {} records failed (retry {}/{})", logger.error("[{}][{}] Still {} records failed (retry {}/{})",
DB_SCHEMA, table, stillFailedIds.size(), record.retryCount, maxRetries); DB_SCHEMA, table, stillFailedIds.size(), record.getRetryCount(), maxRetries);
if (record.retryCount >= maxRetries) { if (record.getRetryCount() >= maxRetries) {
logger.info("[{}][{}] Failed records reached max retries, removing from list", logger.info("[{}][{}] Failed records reached max retries, removing from list",
DB_SCHEMA, table); DB_SCHEMA, table);
failedRecords.tableRecords.remove(table); failedRecords.getTableRecords().remove(table);
} }
} }
} }

2
src/main/java/com/dashboard/aws/lambda/service/MySQLService.java

@ -246,7 +246,7 @@ public class MySQLService {
// 构建IN子句的占位符字符串:?, ?, ?, ... // 构建IN子句的占位符字符串:?, ?, ?, ...
String placeholders = String.join(",", Collections.nCopies(batchIds.size(), "?")); String placeholders = String.join(",", Collections.nCopies(batchIds.size(), "?"));
String sql = String.format("DELETE FROM %s.%s WHERE id IN (%s)", schema, table, placeholders); String sql = String.format("DELETE FROM %s.%s WHERE id IN (%s)", schema, table, placeholders);
logger.info("deleteByIds sql: {}, batch ids count: {}", sql, batchIds.size()); // logger.info("deleteByIds sql: {}, batch ids count: {}", sql, batchIds.size());
try (PreparedStatement ps = conn.prepareStatement(sql)) { try (PreparedStatement ps = conn.prepareStatement(sql)) {
// 逐个设置ID参数值 // 逐个设置ID参数值

Loading…
Cancel
Save