86 lines
2.5 KiB
PHP
86 lines
2.5 KiB
PHP
<?php
|
|
// worker.php
|
|
set_time_limit(0);
|
|
$redisHost = getenv('REDIS_HOST') ?: 'redis';
|
|
$redisPort = getenv('REDIS_PORT') ?: 6379;
|
|
$stream = 'vrfc_requests';
|
|
$group = 'vrfc_group';
|
|
$consumer = 'php_worker_' . bin2hex(random_bytes(3));
|
|
$batchSize = 500;
|
|
|
|
$redis = new Redis();
|
|
$redis->connect($redisHost, $redisPort);
|
|
|
|
// 그룹 생성 (이미 있으면 예외 발생하므로 무시)
|
|
try {
|
|
$redis->xGroup('CREATE', $stream, $group, '$');
|
|
} catch (Exception $e) {}
|
|
|
|
// PDO 연결
|
|
$dsn = sprintf('mysql:host=%s;dbname=%s;charset=utf8mb4', getenv('DB_HOST'), getenv('DB_NAME'));
|
|
$pdo = new PDO($dsn, getenv('DB_USER'), getenv('DB_PASS'), [
|
|
PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
|
|
PDO::ATTR_EMULATE_PREPARES => false,
|
|
]);
|
|
|
|
while (true) {
|
|
$res = $redis->xReadGroup($group, $consumer, [$stream => '>'], $batchSize, 5000);
|
|
if (!$res) {
|
|
// 대기 후 루프
|
|
continue;
|
|
}
|
|
$entries = $res[$stream];
|
|
$ids = [];
|
|
$values = [];
|
|
foreach ($entries as $id => $fields) {
|
|
$ids[] = $id;
|
|
$articleNumbr = $fields['articleNumbr'] ?? '';
|
|
$reqeustType = $fields['reqeustType'] ?? '';
|
|
$requestDatetime = $fields['requestDatetime'] ?? '';
|
|
$values[] = [
|
|
'articleNumbr' => $articleNumbr,
|
|
'reqeustType' => $reqeustType,
|
|
'requestDatetime' => $requestDatetime
|
|
];
|
|
}
|
|
|
|
if (count($values) === 0) {
|
|
continue;
|
|
}
|
|
|
|
// 벌크 인서트 준비 (prepared statements)
|
|
$placeholders = [];
|
|
$params = [];
|
|
foreach ($values as $v) {
|
|
$placeholders[] = "(?, ?, ?)";
|
|
$params[] = $v['articleNumbr'];
|
|
$params[] = $v['reqeustType'];
|
|
$params[] = $v['requestDatetime'];
|
|
}
|
|
$sql = "INSERT INTO vrfc_requests (articleNumbr, reqeustType, requestDatetime) VALUES " . implode(',', $placeholders);
|
|
|
|
try {
|
|
$pdo->beginTransaction();
|
|
$stmt = $pdo->prepare($sql);
|
|
$stmt->execute($params);
|
|
$pdo->commit();
|
|
|
|
// ACK 및 삭제
|
|
foreach ($ids as $id) {
|
|
$redis->xAck($stream, $group, $id);
|
|
$redis->xDel($stream, $id);
|
|
}
|
|
} catch (Exception $e) {
|
|
$pdo->rollBack();
|
|
error_log("DB insert error: " . $e->getMessage());
|
|
// 실패 항목을 실패 스트림으로 이동
|
|
foreach ($entries as $id => $fields) {
|
|
$redis->xAdd('vrfc_failures', '*', 'id', $id, 'payload', json_encode($fields));
|
|
$redis->xAck($stream, $group, $id);
|
|
$redis->xDel($stream, $id);
|
|
}
|
|
// 잠깐 대기
|
|
sleep(2);
|
|
}
|
|
}
|