first commit2

2025-12-16 17:58:37 +09:00
parent 2e3204d5df
commit 515a981cd0
6 changed files with 454 additions and 0 deletions
--- a/.history/worker_20251205190214.php
+++ b/.history/worker_20251205190214.php
@@ -0,0 +1,85 @@
+<?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);
+    }
+}