Google은 AI 기술을 사용하여 콘텐츠를 사용자의 기본 언어로 번역합니다. AI 번역에는 오류가 있을 수 있습니다.

Spanner 대기열 시나리오 및 예

이 문서에서는 Spanner 대기열을 사용하는 일반적인 메시지 시나리오의 아키텍처 패턴과 코드 예시를 제공합니다. 이러한 패턴을 사용하여 트랜잭션이 커밋된 후 비동기 작업을 트리거하고, 지연되거나 반복되는 작업을 예약하고, 대규모 메시지 페이로드를 대역 외 스토리지로 관리하고, 다중 이벤트 워크플로를 조정하고, 장기 실행 백그라운드 작업의 체크포인트를 지정하거나 리스를 연장할 수 있습니다.

정확히 한 번 처리 및 최대 한 번 확인

정확히 한 번 처리 및 최대 한 번 확인에 관한 다양한 고려사항과 해결 방법은 정확히 한 번 처리 및 최대 한 번 확인 페이지에 자세히 설명되어 있습니다.

트랜잭션이 커밋된 후 작업 실행

트랜잭션이 커밋된 후 작업을 실행하려면 동일한 트랜잭션 내에서 큐에 메시지를 전송합니다.

예를 들어 신규 사용자 가입은 환영 이메일을 트리거합니다.

GoogleSQL

-- Inside your application transaction:
-- 1. Insert into Users table
INSERT INTO Users (UserId, UserName) VALUES (124, 'New User');

-- 2. Send message to queue to trigger email
INSERT INTO UserTasks (UserId, MessageId, Payload)
VALUES (
  124,
  'welcome-email-id',
  b'{"type": "welcome", "email": "user@example.com"}'
);

PostgreSQL

-- Inside your application transaction:
-- 1. Insert into users table
INSERT INTO users (userid, username) VALUES (124, 'New User');

-- 2. Send message to queue to trigger email
INSERT INTO usertasks (userid, messageid, payload)
VALUES (
  124,
  'welcome-email-id',
  CAST('{"type": "welcome", "email": "user@example.com"}' AS bytea)
);

트랜잭션이 커밋된 후 UserTasks 수신자는 메시지를 스트리밍하고 이메일을 보내고 메시지를 확인합니다.

GoogleSQL

-- 1. In the receiver process, stream messages from the queue
SELECT
  UserId,
  MessageId,
  Payload,
  DeliverTime,
  SpannerLeaseExpirationTimestamp,
  SpannerLeaseToken
FROM RECEIVE_UserTasks(max_duration=>'20m');

-- 2. After sending the welcome email, acknowledge the message
DELETE FROM UserTasks
WHERE UserId = 124 AND MessageId = 'welcome-email-id';

PostgreSQL

-- 1. In the receiver process, stream messages from the queue
SELECT
  userid,
  messageid,
  payload,
  deliver_time,
  spanner_lease_expiration_timestamp,
  spanner_lease_token
FROM spanner.receive_usertasks(
    max_batch_size=>NULL, priority=>NULL, max_duration=>'20m');

-- 2. After sending the welcome email, acknowledge the message
DELETE FROM usertasks
WHERE userid = 124 AND messageid = 'welcome-email-id';

장기 실행 작업 처리

기본 리스 (10초 초과)보다 오래 걸릴 수 있는 작업이 있는 경우 주기적으로 SELECT * FROM RENEWLEASE_QUEUE_NAME()를 호출합니다.

예를 들어 보고서를 생성하는 경우:

수신자가 RECEIVE_ReportQueue()로부터 메시지를 수신합니다.
보고서 생성을 시작합니다.
5초마다 별도의 스레드나 루틴에서 SELECT * FROM RENEWLEASE_ReportQueue([leaseToken])를 호출합니다.
완료되면 메시지를 확인하고 보고서를 저장합니다.

또는 최대 한 번 처리가 필요하거나 임대 시간이 긴 장기 실행 작업이 있는 경우 다음을 실행하세요.

도착 시 현재 대기열 메시지를 확인합니다 (DELETE 또는 ACK). 동일한 트랜잭션에서 처리하는 데 걸리는 시간을 초과하는 미래의 전송 타임스탬프가 있는 새 대기열 메시지를 다시 대기열에 추가합니다.
처리를 진행하고 완료되면 새로 대기열에 추가된 메시지를 확인합니다.

이 접근 방식의 장점은 리스를 계속 연장할 필요가 없고 미래 시간이 도달할 때까지 메시지가 다시 전송되지 않는다는 것입니다(비정상 종료 포함). 초기 확인이 성공하면 최대 한 번 처리가 달성됩니다.

장기 실행 작업 체크포인트

Spanner 대기열은 빠른 작업뿐만 아니라 몇 분에서 몇 시간 동안 지속되는 작업도 관리할 수 있습니다. 이러한 장기 실행 작업의 경우 다음 방법을 사용하세요.

외부에 메타데이터 저장: 대역 외 스토리지를 사용하여 작업의 세부정보와 상태를 저장합니다.
정기적으로 체크포인트: 다운으로 인한 진행 상황 손실을 최소화하려면 태스크가 주기적으로 상태를 저장해야 합니다.
권장되는 체크포인트 패턴 사용: 체크포인트하는 가장 좋은 방법은 현재 대기열 메시지를 원자적으로 승인 (ACK)하고 향후 전송이 예약된 새 메시지를 보내는 것입니다. 이 새 메시지에는 업데이트된 상태가 포함되거나 이를 가리키므로 다른 작업자에게 즉시 재전송되지 않습니다.

이 패턴은 전체 체크포인트가 불가능한 경우에도 중복 작업을 줄여주지만, 이 시나리오에서는 비정상 종료 후 태스크가 처음부터 다시 시작됩니다.

향후 특정 시간에 작업 예약

향후 특정 시간에 작업을 예약하려면 메시지를 삽입할 때 DeliverTime 열을 설정하세요.

예를 들어 무료 체험 만료 알림은 다음과 같습니다.

GoogleSQL

-- 1. Insert into Users table
INSERT INTO Users (UserId, UserName) VALUES (125, 'Trial User');

-- 2. Send message to queue with a future delivery time
INSERT INTO UserTasks (UserId, MessageId, Payload, DeliverTime)
VALUES (125, 'trial-expire-reminder', b'{"type": "reminder"}', TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 29 DAY));

PostgreSQL

-- 1. Insert into users table
INSERT INTO users (userid, username) VALUES (125, 'Trial User');

-- 2. Send message to queue with a future delivery time
INSERT INTO usertasks (userid, messageid, payload, deliver_time)
VALUES (125, 'trial-expire-reminder', CAST('{"type": "reminder"}' AS bytea), CURRENT_TIMESTAMP + INTERVAL '29 DAY');

대용량 메시지 페이로드 처리

메시지 페이로드가 큰 경우 대역 외 스토리지 패턴을 사용하세요. 큰 페이로드를 별도의 테이블에 저장하고 큐 메시지에 참조를 넣습니다.

예를 들어 이미지 처리의 경우:

GoogleSQL

-- Schema
CREATE TABLE ImageUploads (
  UserId    INT64 NOT NULL,
  ImageId   STRING(36) NOT NULL,
  ImageData BYTES(MAX),
  Status    STRING(MAX) -- PENDING, PROCESSING, DONE
) PRIMARY KEY (UserId, ImageId),
  INTERLEAVE IN PARENT Users;

CREATE QUEUE ImageProcessingQueue (
  UserId    INT64 NOT NULL,
  ImageId   STRING(36) NOT NULL,
  Payload   BYTES(1) NOT NULL -- Payload can be minimal
) PRIMARY KEY (UserId, ImageId),
  INTERLEAVE IN PARENT ImageUploads ON DELETE CASCADE;

-- Application Logic
-- 1. Upload image, insert into ImageUploads with Status 'PENDING'
-- 2. Send message to ImageProcessingQueue
INSERT INTO ImageProcessingQueue (UserId, ImageId, Payload) VALUES (123, 'image-uuid-1', b'');

-- Receiver for ImageProcessingQueue:
-- 1. Receives message (UserId, ImageId).
-- 2. Reads ImageData from ImageUploads.
-- 3. Processes image.
-- 4. Updates ImageUploads Status to 'DONE'.
-- 5. ACKs the queue message.

PostgreSQL

-- Schema
CREATE TABLE imageuploads (
  userid    bigint NOT NULL,
  imageid   varchar(36) NOT NULL,
  imagedata bytea,
  status    varchar, -- PENDING, PROCESSING, DONE
  PRIMARY KEY (userid, imageid)
) INTERLEAVE IN PARENT users;

CREATE QUEUE imageprocessingqueue (
  userid    bigint NOT NULL,
  imageid   varchar(36) NOT NULL,
  payload   bytea NOT NULL, -- Payload can be minimal
  PRIMARY KEY (userid, imageid)
) INTERLEAVE IN PARENT imageuploads ON DELETE CASCADE;

-- Application Logic
-- 1. Upload image, insert into imageuploads with status 'PENDING'
-- 2. Send message to imageprocessingqueue
INSERT INTO imageprocessingqueue (userid, imageid, payload) VALUES (123, 'image-uuid-1', CAST('' AS bytea));

-- Receiver for imageprocessingqueue:
-- 1. Receives message (userid, imageid).
-- 2. Reads imagedata from imageuploads.
-- 3. Processes image.
-- 4. Updates imageuploads status to 'DONE'.
-- 5. ACKs the queue message.

진행하기 전에 여러 이벤트를 기다림

조인 작업과 같이 진행하기 전에 여러 이벤트를 기다리려면 테이블을 사용하여 상태를 추적하고 큐를 사용하여 확인을 트리거하세요.

예를 들어 재고 및 결제가 필요한 주문 처리는 다음과 같습니다.

InventoryStatus 및 PaymentStatus로 Orders 테이블을 만듭니다.
인벤토리가 확인되면 Orders를 업데이트하고 OrderCheckQueue에 메시지를 보냅니다.
결제가 확인되면 Orders를 업데이트하고 OrderCheckQueue에 메시지를 보냅니다.
OrderCheckQueue 수신기는 Orders 테이블을 확인합니다. 두 상태가 모두 확인되면 배송을 진행하고 메시지를 확인합니다. 그렇지 않으면 나중에 다시 확인하도록 대기열에 추가되거나 다른 로직이 실행될 수 있습니다.

주기적으로 작업 실행

주기적으로 작업을 실행하려면 주기적 예약 패턴을 사용하세요. 수신자는 메시지를 확인하고 다음 간격으로 예약된 새 메시지를 보냅니다.

예를 들어 시간별 데이터 집계는 다음과 같습니다.

GoogleSQL

-- Inside your application transaction:
-- 1. Acknowledge current message
DELETE FROM AggregationQueue
WHERE TaskType = 'hourly-aggregator' AND MessageId = 'current-uuid'
ASSERT_ROWS_MODIFIED 1;

-- 2. Schedule next run 1 hour in the future
INSERT INTO AggregationQueue (TaskType, MessageId, Payload, DeliverTime)
VALUES ('hourly-aggregator', 'next-uuid', b'', TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR));

PostgreSQL

-- Inside your application transaction:
-- 1. Acknowledge current message
DELETE FROM aggregationqueue
WHERE tasktype = 'hourly-aggregator' AND messageid = 'current-uuid'
ASSERT_ROWS_MODIFIED 1;

-- 2. Schedule next run 1 hour in the future
INSERT INTO aggregationqueue (tasktype, messageid, payload, deliver_time)
VALUES ('hourly-aggregator', 'next-uuid', CAST('' AS bytea), CURRENT_TIMESTAMP + INTERVAL '1 HOUR');

또는 클라이언트 라이브러리 Ack 및 Send 변형을 사용합니다. 이 예시에서는 키와 페이로드를 캡슐화하는 Message 객체가 있다고 가정합니다.

자바

// Receiver logic for AggregationQueue
public void process(DatabaseClient dbClient, Message msg) {
  // ... do aggregation ...

  // ACK current message and schedule next run (1 hour from now)
  Instant nextRun = Instant.now().plus(Duration.ofHours(1));
  Mutation ackMutation =
      Mutation.newAckBuilder("AggregationQueue")
          .setKey(msg.getKey()) // Ack
          .build();
  Mutation sendMutation =
      Mutation.newSendBuilder("AggregationQueue")
          .setKey(Key.of("hourly-aggregator", "next-uuid"))
          .setPayload(Value.bytes(ByteArray.copyFrom("")))
          .setDeliveryTime(nextRun) // Schedule next
          .build();
  dbClient.write(Arrays.asList(ackMutation, sendMutation));
}

Go

// Receiver logic for AggregationQueue
func process(msg) {
    // ... do aggregation ...

    // ACK current message and schedule next run
    nextRun := time.Now().Add(1 * time.Hour)
    _, err := client.Apply(ctx, []*spanner.Mutation{
        spanner.Ack("AggregationQueue", msg.Key), // Ack
        spanner.Send("AggregationQueue",
            spanner.Key{"hourly-aggregator", "next-uuid"},
            []byte(""),
            spanner.WithDeliveryTime(nextRun), // Schedule next
        ),
    })
    // ... handle err ...
}

Python

# Receiver logic for AggregationQueue
def process(database: spanner.Database, msg: Message):
  # ... do aggregation ...
  # ACK current message and schedule next run (1 hour from now)
  next_run = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
      hours=1
  )
  with database.batch() as batch:
    batch.ack(
        queue="AggregationQueue",
        key=msg.key,  # Ack
    )
    batch.send(
        queue="AggregationQueue",
        key=("hourly-aggregator", "next-uuid"),
        payload=b"",
        deliver_time=next_run,  # Schedule next
    )

Node.js

/**
 * Receiver logic for AggregationQueue
 * @param {import('@google-cloud/spanner').Database} database
 * @param { { key: Array<string|number>, payload: Buffer } } msg
 */
async function process(database, msg) {
  // ... do aggregation ...
  // ACK current message and schedule next run (1 hour from now)
  const nextRun = new Date(Date.now() + 60 * 60 * 1000);
  await database.runTransactionAsync(async (transaction) => {
    // Ack current message
    transaction.queueAck('AggregationQueue', msg.key);
    // Schedule next run
    transaction.queueSend(
      'AggregationQueue',
      ['hourly-aggregator', 'next-uuid'],
      {
        payload: Buffer.from(''),
        deliverTime: nextRun,
      }
    );
    await transaction.commit();
  });
}

시간 일괄 처리를 사용하여 메시지 일괄 처리

Spanner 대기열은 최소한의 지연 시간으로 메시지를 전송합니다. 하지만 독립적인 클라이언트에서 많은 양의 메시지가 지속적으로 도착하는 경우 각 메시지를 개별적으로 처리하면 트랜잭션 오버헤드가 높아질 수 있습니다. 메시지를 일괄 처리하기 위해 대기열 테이블을 수동으로 쿼리하거나 스캔하면 범위 잠금 경합, 높은 중단 비율, 추가 읽기 비용이 발생할 수 있습니다.

경합 없이 높은 처리량의 일괄 처리를 달성하려면 시간적 일괄 처리 패턴을 적용하세요. 전송자는 메시지의 DeliverTime를 미래의 불연속 시간 창에 정렬합니다 (예: 가장 가까운 10초 경계로 반올림). 독립적인 발신자가 동일한 미래 타임스탬프를 계산하므로 Spanner는 동일한 분할의 메시지를 함께 그룹화하고 RECEIVE_QUEUE_NAME() 테이블 값 함수의 max_batch_size가 허용하는 경우 단일 배치로 전송하거나, max_batch_size가 전송할 메시지 수보다 작은 경우 여러 배치로 전송하는 경향이 있습니다.

전송 타임스탬프는 다음 수식으로 계산할 수 있습니다.

$$ \text{DeliverTime} = \text{RoundDown}(\text{CurrentTime}, \text{FixedDelay}) + \text{FixedDelay} $$

예를 들어 10초 윈도우를 사용하면 09:05:00과 09:05:09.999 사이에 대기열에 추가된 메시지가 모두 09:05:10의 DeliverTime를 수신합니다.

GoogleSQL

-- Calculate delivery time rounded to the next 10-second interval.
-- Use DIV(..., 10) * 10 to perform the RoundDown in SQL:
INSERT INTO OrderProcessingQueue (OrderId, DeliverTime, Payload)
VALUES (
  'order-101',
  TIMESTAMP_SECONDS(DIV(UNIX_SECONDS(CURRENT_TIMESTAMP()), 10) * 10 + 10),
  b'{"item": "book", "qty": 1}'
);

PostgreSQL

-- Calculate delivery time rounded to the next 10-second interval:
INSERT INTO orderprocessingqueue (orderid, deliver_time, payload)
VALUES (
  'order-101',
  to_timestamp((floor(extract(epoch from CURRENT_TIMESTAMP) / 10) * 10) + 10),
  CAST('{"item": "book", "qty": 1}' AS bytea)
);

그런 다음 수신자는 max_batch_size를 사용하여 이러한 동기화된 메시지를 일괄적으로 가져옵니다.

GoogleSQL

SELECT OrderId, Payload, DeliverTime, SpannerLeaseToken
FROM RECEIVE_OrderProcessingQueue(max_duration=>'20m', max_batch_size=>50);

PostgreSQL

SELECT orderid, payload, deliver_time, spanner_lease_token
FROM spanner.receive_orderprocessingqueue(
    max_batch_size=>50, priority=>NULL, max_duration=>'20m');

수백만 개의 메시지를 동시에 보내는 경우 모든 메시지를 정확히 동일한 초에 정렬하면 갑작스러운 처리 급증이 발생할 수 있습니다. 작업을 균등하게 분배하면서도 엔티티별로 메시지를 일괄처리하려면 고유 식별자 (예: TIMESTAMP_ADD(deliver_time, INTERVAL MOD(ABS(FARM_FINGERPRINT(CAST(UserId AS STRING))), 10) SECOND))를 기반으로 계산에 오프셋을 추가합니다.

처리에서 데이터 수정 분리 (더티 플래그 패턴)

처리량이 많은 트랜잭션 애플리케이션에서 사용자 대상 트랜잭션 내에서 직접 복잡한 재계산, 검색 색인 생성 또는 캐시 무효화를 실행하면 지연 시간이 증가하고 공유 행에서 잠금 경합이 발생하여 사용자 환경이 느려질 수 있습니다.

더티 플래그 패턴은 데이터 수정사항을 비동기 처리에서 분리합니다. 트랜잭션이 테이블을 수정하면 동일한 트랜잭션 내의 대기열에 경량 '더티 비트' 메시지를 작성합니다. 그런 다음 백그라운드 작업자가 메시지를 소비하고 비용이 많이 드는 처리를 비동기적으로 실행합니다.

예를 들어 고객이 프로필이나 설정을 변경하는 경우:

GoogleSQL

-- Inside user profile update transaction:
-- 1. Update the primary entity table
UPDATE UserProfiles
SET FullName = 'Jane Doe', UpdatedAt = CURRENT_TIMESTAMP()
WHERE UserId = 456;

-- 2. Send lightweight dirty flag message to the queue.
-- It is recommended to interleave the queue in the primary UserProfiles
-- table for better transaction performance.
INSERT INTO UserDirtyQueue (UserId, TaskType, CommitTimestamp, Payload)
VALUES (456, 'reindex-user-profile', CURRENT_TIMESTAMP(), b'');

PostgreSQL

-- Inside user profile update transaction:
-- 1. Update the primary entity table
UPDATE userprofiles
SET fullname = 'Jane Doe', updatedat = CURRENT_TIMESTAMP
WHERE userid = 456;

-- 2. Send lightweight dirty flag message to the queue
INSERT INTO userdirtyqueue (userid, tasktype, committimestamp, payload)
VALUES (456, 'reindex-user-profile', CURRENT_TIMESTAMP, CAST('' AS bytea));

UserDirtyQueue의 백그라운드 수신기는 UserId를 수신하고, 사용자의 중요 경로 외부에서 최신 프로필 행을 읽고, 검색 색인을 다시 계산하거나 외부 캐시를 업데이트합니다. 짧은 기간 내에 동일한 사용자에 대한 여러 업데이트가 큐로 전송되는 경우 이 패턴에도 일괄 처리를 적용할 수 있습니다. 이 경우 수신기 TVF는 동일한 배치에서 여러 메시지를 수신하기 위해 1보다 큰 max_batch_size를 지정할 수 있습니다.

작업자 상태 모니터링 및 타임아웃 감지

예약된 대기열 메시지를 사용하여 작업자 노드 또는 마이크로서비스 인스턴스 집단의 내결함성 하트비트 및 상태 모니터링 시스템을 빌드할 수 있습니다.

상태 점검을 구현하려면 다음 단계를 따르세요.

시작 시 작업자 등록: 작업자가 초기화되면 실패 기한 (예: 60초)으로 설정된 미래 DeliverTime와 함께 상태 점검 대기열에 하트비트 메시지를 삽입합니다.
주기적인 하트비트 전송: 정상 상태인 동안 작업자는 DeliverTime를 60초 더 앞으로 이동하여 주기적으로 (예: 10초마다) 하트비트 메시지를 새로고침합니다.
실패 감지: 작업자가 비정상 종료되거나 네트워크 연결이 끊어지면 하트비트 새로고침이 중지됩니다. 60초가 지나면 전송 타임스탬프가 만료되고(DeliverTime <= CURRENT_TIMESTAMP()) Spanner가 알림 수신자에게 메시지를 전송하여 장애 조치 또는 작업 재할당을 시작합니다.

중요: Spanner 큐는 UPDATE DML 문을 지원하지 않습니다. 따라서 하트비트 타임스탬프를 새로고침하려면 기존 메시지를 삭제하고 단일 트랜잭션 내에서 새 DeliverTime로 대체 항목을 삽입하거나 클라이언트 라이브러리 Ack 및 Send 변형을 적용해야 합니다. 대기열의 기본 키가 (WorkerId, MessageId)이 아닌 WorkerId 단독인지 확인하여 특정 시점에 작업자당 하나의 하트비트 메시지만 존재하도록 합니다.

GoogleSQL

-- Inside the worker heartbeat transaction (executed every 10 seconds):
-- 1. Acknowledge the existing heartbeat message
DELETE FROM WorkerHealthQueue
WHERE WorkerId = 'worker-node-42' ASSERT_ROWS_MODIFIED 1;

-- 2. Send replacement heartbeat with refreshed 60-second deadline
INSERT INTO WorkerHealthQueue (WorkerId, DeliverTime, Payload)
VALUES (
  'worker-node-42',
  TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 60 SECOND),
  b'{"status": "healthy", "active_jobs": 3}'
);

PostgreSQL

-- Inside the worker heartbeat transaction (executed every 10 seconds):
-- 1. Acknowledge the existing heartbeat message
DELETE FROM workerhealthqueue
WHERE workerid = 'worker-node-42' ASSERT_ROWS_MODIFIED 1;

-- 2. Send replacement heartbeat with refreshed 60-second deadline
INSERT INTO workerhealthqueue (workerid, deliver_time, payload)
VALUES (
  'worker-node-42',
  CURRENT_TIMESTAMP + INTERVAL '60 SECOND',
  CAST('{"status": "healthy", "active_jobs": 3}' AS bytea)
);

클라이언트 라이브러리 변경사항을 사용하는 경우 이 Go 예와 같이 변경사항 슬라이스에서 Ack이 Send보다 앞에 있어야 합니다.

// Worker heartbeat loop
func sendHeartbeat(ctx context.Context, client *spanner.Client, workerID string) error {
    newDeadline := time.Now().Add(60 * time.Second)
    _, err := client.Apply(ctx, []*spanner.Mutation{
        spanner.Ack("WorkerHealthQueue", spanner.Key{workerID}),
        spanner.Send(
            "WorkerHealthQueue",
            spanner.Key{workerID},
            []byte(`{"status":"healthy"}`),
            spanner.WithDeliveryTime(newDeadline),
        ),
    })
    return err
}

작업자가 정상적으로 종료되면 잘못된 알림이 트리거되지 않도록 하트비트 메시지를 명시적으로 삭제합니다.

DELETE FROM WorkerHealthQueue WHERE WorkerId = 'worker-node-42' ASSERT_ROWS_MODIFIED 1;

단일 대기열에서 여러 작업 유형 처리

Spanner 인스턴스의 총 대기열 수에는 한도가 있습니다. 작은 비동기 작업마다 별도의 대기열을 만들면 이 한도에 빠르게 도달할 수 있으며 동시 수신자 쿼리를 많이 관리해야 합니다.

작업을 통합하려면 다양한 작업 유형을 다형성 대기열이라는 단일 대기열로 결합하세요. 다형 큐를 만드는 데 사용되는 전략에는 두 가지가 있습니다.

전략 1: 기본 키에 유형 열 포함

GoogleSQL

CREATE QUEUE ApplicationTasks (
  TaskType   STRING(50) NOT NULL,
  TaskId     STRING(36) NOT NULL,
  Payload    BYTES(MAX) NOT NULL,
) PRIMARY KEY (TaskType, TaskId);

-- Enqueue an email task
INSERT INTO ApplicationTasks (TaskType, TaskId, Payload)
VALUES ('SEND_EMAIL', 'task-uuid-1', b'{"to": "user@example.com", "template": "welcome"}');

-- Enqueue an image thumbnail task
INSERT INTO ApplicationTasks (TaskType, TaskId, Payload)
VALUES ('GENERATE_THUMBNAIL', 'task-uuid-2', b'{"image_id": "img-789", "size": "small"}');

PostgreSQL

CREATE QUEUE applicationtasks (
  tasktype   varchar(50) NOT NULL,
  taskid     varchar(36) NOT NULL,
  payload    bytea NOT NULL,
  PRIMARY KEY (tasktype, taskid)
);

-- Enqueue an email task
INSERT INTO applicationtasks (tasktype, taskid, payload)
VALUES ('SEND_EMAIL', 'task-uuid-1', CAST('{"to": "user@example.com", "template": "welcome"}' AS bytea));

-- Enqueue an image thumbnail task
INSERT INTO applicationtasks (tasktype, taskid, payload)
VALUES ('GENERATE_THUMBNAIL', 'task-uuid-2', CAST('{"image_id": "img-789", "size": "small"}' AS bytea));

수신자는 TaskType를 검사하고 페이로드를 해당 핸들러에 디스패치합니다.

전략 2: 다형성 페이로드 구조

또는 작업 또는 유형 식별자 필드가 포함된 JSON 페이로드를 사용합니다.

{
  "action": "SYNC_INVENTORY",
  "data": { "item_id": 987, "delta": -1 }
}

맞춤 재시도 지연 구현

Spanner 대기열은 실패하거나 확인되지 않은 메시지를 내장된 지수 백오프로 자동 재시도합니다. 하지만 알려진 기간이 있는 알려진 이유로 메시지가 실패하거나 외부 비율 제한 (예: Retry-After 헤더를 지정하는 HTTP 429 응답)이 있는 시나리오에서는 자동 백오프에 의존하면 CPU 리소스를 낭비하는 조기 재시도 시도가 발생할 수 있습니다.

맞춤 재시도 지연을 구현하려면 다음을 실행하세요.

메시지 프로세서에서 특정 일시적 오류를 포착합니다.
현재 전송 시도를 충족하기 위해 현재 메시지를 확인합니다.
동일한 트랜잭션에서 선택한 향후 재시도 시간 (다음 예에서는 5분 후)으로 명시적으로 설정된 DeliverTime를 사용하여 대체 메시지를 전송합니다.

GoogleSQL

-- Inside failure-handling transaction:
-- 1. Acknowledge the failed message
DELETE FROM OutboundNotificationQueue
WHERE NotificationId = 'notif-555' ASSERT_ROWS_MODIFIED 1;

-- 2. Reschedule delivery 5 minutes in the future
INSERT INTO OutboundNotificationQueue (NotificationId, DeliverTime, Payload)
VALUES (
  'notif-555',
  TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 5 MINUTE),
  b'{"recipient": "user@example.com", "retry_count": 2}'
);

PostgreSQL

-- Inside failure-handling transaction:
-- 1. Acknowledge the failed message
DELETE FROM outboundnotificationqueue
WHERE notificationid = 'notif-555' ASSERT_ROWS_MODIFIED 1;

-- 2. Reschedule delivery 5 minutes in the future
INSERT INTO outboundnotificationqueue (notificationid, deliver_time, payload)
VALUES (
  'notif-555',
  CURRENT_TIMESTAMP + INTERVAL '5 MINUTE',
  CAST('{"recipient": "user@example.com", "retry_count": 2}' AS bytea)
);

이 방식을 사용하면 애플리케이션에서 백오프 일정을 정확하게 관리하고 다운스트림 복구 기간 동안 외부 API가 포화 상태가 되는 것을 방지할 수 있습니다.

다음 단계

권장사항 및 모니터링을 비롯한 Spanner 대기열 사용 방법을 알아봅니다.
정확히 한 번 처리 및 최대 한 번 확인에 대해 알아봅니다.
대기열의 세분화된 액세스 제어로 액세스 제어를 구성합니다.

Spanner 대기열 시나리오 및 예 컬렉션을 사용해 정리하기 내 환경설정을 기준으로 콘텐츠를 저장하고 분류하세요.

정확히 한 번 처리 및 최대 한 번 확인

트랜잭션이 커밋된 후 작업 실행

GoogleSQL

PostgreSQL

GoogleSQL

PostgreSQL

장기 실행 작업 처리

장기 실행 작업 체크포인트

향후 특정 시간에 작업 예약

GoogleSQL

PostgreSQL

대용량 메시지 페이로드 처리

GoogleSQL

PostgreSQL

진행하기 전에 여러 이벤트를 기다림

주기적으로 작업 실행

GoogleSQL

PostgreSQL

자바

Go

Python

Node.js

시간 일괄 처리를 사용하여 메시지 일괄 처리

GoogleSQL

PostgreSQL

GoogleSQL

PostgreSQL

처리에서 데이터 수정 분리 (더티 플래그 패턴)

GoogleSQL

PostgreSQL

작업자 상태 모니터링 및 타임아웃 감지

GoogleSQL

PostgreSQL

단일 대기열에서 여러 작업 유형 처리

전략 1: 기본 키에 유형 열 포함

GoogleSQL

PostgreSQL

전략 2: 다형성 페이로드 구조

맞춤 재시도 지연 구현

GoogleSQL

PostgreSQL

다음 단계

Spanner 대기열 시나리오 및 예