Skenario dan contoh antrean Spanner

Dokumen ini memberikan pola arsitektur dan contoh kode untuk skenario pengiriman pesan umum menggunakan antrean Spanner. Anda dapat menggunakan pola ini untuk memicu tugas asinkron setelah transaksi di-commit, menjadwalkan tugas yang tertunda atau berulang, mengelola payload pesan besar dengan penyimpanan di luar band, mengoordinasikan alur kerja multi-peristiwa, dan membuat checkpoint atau memperpanjang masa berlaku untuk tugas latar belakang yang berjalan lama.

Pemrosesan tepat satu kali dan pengakuan paling banyak satu kali

Berbagai pertimbangan dan solusi untuk pemrosesan tepat satu kali dan pengakuan paling banyak satu kali dijelaskan secara lebih mendetail di halaman Pemrosesan tepat satu kali dan pengakuan paling banyak satu kali.

Melakukan pekerjaan setelah transaksi di-commit

Untuk melakukan pekerjaan setelah transaksi di-commit, kirim pesan ke antrean dalam transaksi yang sama.

Misalnya, pendaftaran pengguna baru memicu email sambutan:

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)
);

Setelah transaksi di-commit, penerima untuk UserTasks melakukan streaming pesan, mengirim email, dan mengonfirmasi pesan:

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';

Menangani pekerjaan yang berjalan lama

Jika Anda memiliki pekerjaan yang mungkin memerlukan waktu lebih lama dari masa sewa default (lebih dari 10 detik), panggil SELECT * FROM RENEWLEASE_QUEUE_NAME() secara berkala.

Misalnya, membuat laporan:

  1. Penerima akan mendapatkan pesan dari RECEIVE_ReportQueue().
  2. Mulai pembuatan laporan.
  3. Setiap 5 detik, panggil SELECT * FROM RENEWLEASE_ReportQueue([leaseToken]) dalam thread atau rutin terpisah.
  4. Setelah selesai, konfirmasi pesan dan simpan laporan.

Atau, jika Anda memiliki pekerjaan yang berjalan lama yang memerlukan pemrosesan paling banyak sekali, atau waktu sewa yang lama, lakukan hal berikut:

  1. Mengonfirmasi (DELETE atau ACK) pesan antrean saat ini saat tiba. Dalam transaksi yang sama, antrekan ulang pesan antrean baru dengan stempel waktu pengiriman di masa mendatang, di luar waktu yang diperlukan untuk pemrosesan.
  2. Lanjutkan pemrosesan, dan konfirmasi pesan yang baru dimasukkan ke dalam antrean setelah selesai.

Keuntungan dari pendekatan ini adalah tidak perlu terus memperpanjang masa sewa, dan pesan tidak dikirim ulang hingga waktu mendatang tiba (yang mencakup error). Jika pengakuan awal berhasil, maka akan tercapai pemrosesan paling banyak sekali.

Membuat checkpoint untuk pekerjaan yang berjalan lama

Antrean Spanner dapat mengelola tugas yang berlangsung dari beberapa menit hingga beberapa jam, bukan hanya tugas cepat. Untuk tugas yang berjalan lama ini, gunakan pendekatan berikut:

  1. Simpan metadata secara eksternal: Gunakan penyimpanan di luar band untuk menyimpan detail dan status tugas.
  2. Lakukan checkpoint secara rutin: Untuk memulihkan dari error tanpa kehilangan banyak progres, tugas harus menyimpan statusnya secara berkala.
  3. Gunakan pola pembuatan titik pemeriksaan yang direkomendasikan: Cara terbaik untuk membuat titik pemeriksaan adalah dengan mengonfirmasi (ACK) pesan antrean saat ini secara atomik dan mengirim pesan baru yang dijadwalkan untuk pengiriman pada masa mendatang. Pesan baru ini berisi atau mengarah ke status yang diperbarui, yang mencegah pengiriman ulang langsung ke pekerja lain.

Pola ini mengurangi pekerjaan duplikat meskipun checkpointing penuh tidak memungkinkan, meskipun tugas dimulai ulang dari awal setelah terjadi error dalam skenario tersebut.

Menjadwalkan pekerjaan untuk waktu tertentu pada masa mendatang

Untuk menjadwalkan pekerjaan pada waktu tertentu di masa mendatang, tetapkan kolom DeliverTime saat menyisipkan pesan.

Misalnya, pengingat waktu berakhir uji coba:

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');

Menangani payload pesan besar

Jika payload pesan Anda besar, gunakan pola penyimpanan di luar band. Simpan payload besar dalam tabel terpisah dan masukkan referensinya ke dalam pesan antrean.

Misalnya, pemrosesan gambar:

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.

Menunggu beberapa peristiwa sebelum melanjutkan

Untuk menunggu beberapa peristiwa sebelum melanjutkan (seperti operasi gabungan), gunakan tabel untuk melacak status dan antrean untuk memicu pemeriksaan.

Misalnya, pemenuhan pesanan yang memerlukan inventaris dan pembayaran:

  1. Buat tabel Orders dengan InventoryStatus dan PaymentStatus.
  2. Setelah inventaris dikonfirmasi, perbarui Orders dan kirim pesan ke OrderCheckQueue.
  3. Setelah pembayaran dikonfirmasi, perbarui Orders dan kirim pesan ke OrderCheckQueue.
  4. Penerima untuk OrderCheckQueue memeriksa tabel Orders. Jika kedua status dikonfirmasi, penjual akan melanjutkan pengiriman dan mengonfirmasi pesan. Jika tidak, pemeriksaan mungkin dijadwalkan ulang untuk pemeriksaan nanti atau menjalankan logika lain.

Melakukan tindakan secara berkala

Untuk melakukan tindakan secara berkala, gunakan pola penjadwalan berkala. Penerima mengonfirmasi pesan dan mengirim pesan baru yang dijadwalkan untuk interval berikutnya.

Misalnya, agregasi data per jam:

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');

Atau, gunakan mutasi Ack dan Send library klien. Contoh ini mengasumsikan Anda memiliki objek Message yang merangkum kunci dan payload:

Java 

// 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();
  });
}

Langkah berikutnya