使用 Edge Functions 消费 Supabase 队列消息

Learn how to consume Supabase Queue messages server-side with a Supabase Edge Function

本指南帮助你使用 Supabase Edge Function 在服务器端读取和处理队列消息。想了解我们 API 的更多细节，请阅读 Queues API 参考。

🌐 This guide helps you read & process queue messages server-side with a Supabase Edge Function. Read Queues API Reference for more details on our API.

概念 #

🌐 Concepts

Supabase Queues 是一个基于拉取的消息队列，主要由三个组件组成：队列、消息和队列类型。你应该已经熟悉队列快速入门。

🌐 Supabase Queues is a pull-based Message Queue consisting of three main components: Queues, Messages, and Queue Types. You should already be familiar with the Queues Quickstart.

在 Edge 函数中消费消息 #

🌐 Consuming messages in an Edge Function

这是一个 Supabase Edge 函数，它会从队列中读取 5 条消息，处理每条消息，并在处理完成后删除每条消息。

🌐 This is a Supabase Edge Function that reads 5 messages off the queue, processes each of them, and deletes each message when it is done.

1
import 'jsr:@supabase/functions-js/edge-runtime.d.ts'
2
3
import { createClient } from 'npm:@supabase/supabase-js@2'
4
5
const supabaseUrl = 'supabaseURL'
6
const supabaseKey = 'supabaseKey'
7
8
const supabase = createClient(supabaseUrl, supabaseKey)
9
const queueName = 'your_queue_name'
10
11
// Type definition for queue messages
12
interface QueueMessage {
13
  msg_id: bigint
14
  read_ct: number
15
  vt: string
16
  enqueued_at: string
17
  message: any
18
}
19
20
async function processMessage(message: QueueMessage) {
21
  //
22
  // Do whatever logic you need to with the message content
23
  //
24
  // Delete the message from the queue
25
  const { error: deleteError } = await supabase.schema('pgmq_public').rpc('delete', {
26
    queue_name: queueName,
27
    msg_id: message.msg_id,
28
  })
29
30
  if (deleteError) {
31
    console.error(`Failed to delete message ${message.msg_id}:`, deleteError)
32
  } else {
33
    console.log(`Message ${message.msg_id} deleted from queue`)
34
  }
35
}
36
37
Deno.serve(async (req) => {
38
  const { data: messages, error } = await supabase.schema('pgmq_public').rpc('read', {
39
    queue_name: queueName,
40
    sleep_seconds: 0, // Don't wait if queue is empty
41
    n: 5, // Read 5 messages off the queue
42
  })
43
44
  if (error) {
45
    console.error(`Error reading from ${queueName} queue:`, error)
46
    return new Response(JSON.stringify({ error: error.message }), {
47
      status: 500,
48
      headers: { 'Content-Type': 'application/json' },
49
    })
50
  }
51
52
  if (!messages || messages.length === 0) {
53
    console.log('No messages in workflow_messages queue')
54
    return new Response(JSON.stringify({ message: 'No messages in queue' }), {
55
      status: 200,
56
      headers: { 'Content-Type': 'application/json' },
57
    })
58
  }
59
60
  console.log(`Found ${messages.length} messages to process`)
61
62
  // Process each message that was read off the queue
63
  for (const message of messages) {
64
    try {
65
      await processMessage(message as QueueMessage)
66
    } catch (error) {
67
      console.error(`Error processing message ${message.msg_id}:`, error)
68
    }
69
  }
70
71
  // Return immediately while background processing continues
72
  return new Response(
73
    JSON.stringify({
74
      message: `Processing ${messages.length} messages in background`,
75
      count: messages.length,
76
    }),
77
    {
78
      status: 200,
79
      headers: { 'Content-Type': 'application/json' },
80
    }
81
  )
82
})

每次运行这个 Edge 函数时，它会：

🌐 Every time this Edge Function is run it:

从队列中读取5条消息
调用 processMessage 函数
processMessage结束时，消息会从队列中删除
如果 processMessage 抛出错误，错误会被记录下来。在这种情况下，消息仍然在队列中，所以下一次这个 Edge Function 运行时，它会再次读取这条消息。

你可能会发现这种设置在配合 Supabase Cron 使用时很方便。你可以设置 Cron，让每隔 N 分钟或秒，Edge Function 就会运行并处理队列中的一些消息。

🌐 You might find this kind of setup handy to run with Supabase Cron. You can set up Cron so that every N number of minutes or seconds, the Edge Function will run and process a number of messages off the queue.

同样，你可以在任何时候通过 supabase.functions.invoke 命令调用 Edge 函数。

🌐 Similarly, you can invoke the Edge Function on command at any given time with supabase.functions.invoke.