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(NULL, NULL, '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 대기열을 사용하는 방법을 알아봅니다.
정확히 한 번 처리 및 최대 한 번 확인에 대해 알아봅니다.
대기열에 대한 세분화된 액세스 제어로 액세스 제어를 구성합니다.

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

단 한 번 처리 및 최대 한 번 확인

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

GoogleSQL

PostgreSQL

GoogleSQL

PostgreSQL

장기 실행 작업 처리

장기 실행 작업 체크포인트

향후 특정 시간에 작업 예약

GoogleSQL

PostgreSQL

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

GoogleSQL

PostgreSQL

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

주기적으로 작업 실행

GoogleSQL

PostgreSQL

자바

Go

Python

Node.js

다음 단계

Spanner 대기열 시나리오 및 예