SimpleJdbc 에 batchUpdate(String sql, List<Object[]> argsList, int batchSize) 를 추가하세요. 하나의 트랜잭션 안에서 addBatch() 를 쌓다가 batchSize 마다 executeBatch() 하고, 꼬리를 처리한 뒤 커밋해 총 영향 행 수를 반환합니다. 예외 시 롤백합니다.
public int batchUpdate(String sql, List<Object[]> argsList, int batchSize) {
return inTransaction(c -> {
int total = 0;
try (PreparedStatement ps = c.prepareStatement(sql)) {
int pending = 0;
for (Object[] args : argsList) {
bind(ps, args);
ps.addBatch();
if (++pending == batchSize) {
for (int n : ps.executeBatch()) total += Math.max(n, 0); // SUCCESS_NO_INFO(-2) 는 0 으로
pending = 0;
}
}
if (pending > 0) for (int n : ps.executeBatch()) total += Math.max(n, 0);
}
return total;
});
}
// 사용
List<Object[]> rows = new ArrayList<>();
for (int i = 0; i < 10_000; i++) rows.add(new Object[]{i, "row-" + i});
int affected = jdbc.batchUpdate("INSERT INTO bulk_log(seq, body) VALUES (?, ?)", rows, 1000);
System.out.println(affected); // 10000OrderService.placeOrder 를 확장해 회원의 balance 로 결제하는 payMethod = "BALANCE" 를 지원하세요.
회원 행을 FOR UPDATE 로 잠그고 잔액이 부족하면 InsufficientBalanceException 을 던져 재고 차감까지 롤백되게 하고, 충분하면 balance 를 차감한 뒤 결제 행을 넣습니다.
성공 후 member.balance 와 product.stock 이 함께 줄었는지, 실패 시 둘 다 그대로인지 확인하는 코드까지 작성하세요.
public static class InsufficientBalanceException extends RuntimeException {
public InsufficientBalanceException(String m) { super(m); }
}
public long placeOrder(long memberId, long productId, int qty, String payMethod) {
return jdbc.inTransaction(c -> {
List<Object[]> rows = SimpleJdbc.query(c, "SELECT price, stock FROM product WHERE id = ? FOR UPDATE",
rs -> new Object[]{rs.getBigDecimal("price"), rs.getInt("stock")}, productId);
if (rows.isEmpty()) throw new IllegalArgumentException("상품 없음: " + productId);
BigDecimal price = (BigDecimal) rows.get(0)[0];
int stock = (int) rows.get(0)[1];
if (stock < qty) throw new OutOfStockException("재고 부족");
SimpleJdbc.update(c, "UPDATE product SET stock = stock - ? WHERE id = ?", qty, productId);
BigDecimal amount = price.multiply(BigDecimal.valueOf(qty));
if ("BALANCE".equals(payMethod)) {
BigDecimal balance = SimpleJdbc.query(c, "SELECT balance FROM member WHERE id = ? FOR UPDATE",
rs -> rs.getBigDecimal(1), memberId).stream().findFirst()
.orElseThrow(() -> new IllegalArgumentException("회원 없음: " + memberId));
if (balance.compareTo(amount) < 0)
throw new InsufficientBalanceException("잔액 " + balance.toPlainString() + " < 주문액 " + amount.toPlainString());
SimpleJdbc.update(c, "UPDATE member SET balance = balance - ? WHERE id = ?", amount, memberId);
}
long orderId = SimpleJdbc.insertReturningKey(c,
"INSERT INTO orders(member_id, product_id, qty, amount, status) VALUES (?,?,?,?,'PAID')", memberId, productId, qty, amount);
SimpleJdbc.update(c, "INSERT INTO payment(order_id, amount, method) VALUES (?,?,?)", orderId, amount, payMethod);
return orderId;
});
}
// 확인
RowMapper<String> state = rs -> "balance=" + rs.getBigDecimal(1).toPlainString() + " stock=" + rs.getInt(2);
String q = "SELECT m.balance, p.stock FROM member m, product p WHERE m.id = 1 AND p.id = 2";
System.out.println(jdbc.query(q, state).get(0)); // balance=150000.50 stock=100
svc.placeOrder(1, 2, 2, "BALANCE"); // 마우스 2개 = 50000
System.out.println(jdbc.query(q, state).get(0)); // balance=100000.50 stock=98
try { svc.placeOrder(1, 2, 5, "BALANCE"); } // 125000 > 100000.50
catch (OrderService.InsufficientBalanceException e) { System.out.println(e.getMessage()); }
System.out.println(jdbc.query(q, state).get(0)); // balance=100000.50 stock=98 (재고도 원복)키셋 페이징으로 member 전체를 200 건씩 순회하며 grade 를 재산정하는 배치를 작성하세요. 페이지마다 하나의 트랜잭션으로 UPDATE 를 배치 실행하고 커밋하며, 마지막으로 처리한 id 를 batch_checkpoint(job_name, last_id) 테이블에 같은 트랜잭션으로 기록해 재시작 시 이어서 처리되게 합니다(배치 04 레슨 2.4 의 "같은 DB 체크포인트").
// CREATE TABLE batch_checkpoint(job_name VARCHAR(50) PRIMARY KEY, last_id BIGINT NOT NULL);
static String gradeOf(BigDecimal balance) {
if (balance.compareTo(new BigDecimal("100000")) >= 0) return "GOLD";
if (balance.compareTo(new BigDecimal("10000")) >= 0) return "SILVER";
return "BRONZE";
}
static void regradeAll(SimpleJdbc jdbc, String jobName, int pageSize) {
long lastId = jdbc.queryOne("SELECT last_id FROM batch_checkpoint WHERE job_name = ?", rs -> rs.getLong(1), jobName).orElse(0L);
if (lastId > 0) System.out.println("[restart] id " + lastId + " 이후부터");
while (true) {
final long from = lastId;
Long next = jdbc.inTransaction(c -> {
List<MemberDao.Member> page = SimpleJdbc.query(c,
"SELECT * FROM member WHERE id > ? ORDER BY id LIMIT ?", MemberDao.MAPPER, from, pageSize);
if (page.isEmpty()) return null;
try (PreparedStatement ps = c.prepareStatement("UPDATE member SET grade = ? WHERE id = ?")) {
for (MemberDao.Member m : page) { ps.setString(1, gradeOf(m.balance())); ps.setLong(2, m.id()); ps.addBatch(); }
ps.executeBatch();
}
long last = page.get(page.size() - 1).id();
SimpleJdbc.update(c, "MERGE INTO batch_checkpoint(job_name, last_id) KEY(job_name) VALUES (?, ?)", jobName, last); // 데이터와 같은 tx
return last;
});
if (next == null) break;
lastId = next;
System.out.println("페이지 커밋: last_id=" + lastId);
}
jdbc.update("DELETE FROM batch_checkpoint WHERE job_name = ?", jobName); // 정상 종료: 다음 실행은 처음부터
}체크포인트 MERGE 가 UPDATE 들과 같은 commit() 에 묶이므로 "커밋됐는데 체크포인트가 없는" 창이 없습니다. grade = calc(balance) 는 절대값 갱신이라 재실행에도 멱등합니다.