Supabase实时订阅概述 #

Supabase Realtime
├── 基于WebSocket的实时通信
├── 数据库变更监听
├── 广播消息
├── 在线状态
└── 低延迟实时更新

┌─────────────────────────────────────────────────────────────┐
│                        客户端应用                            │
│         Web / Mobile / Desktop                              │
└─────────────────────────────────────────────────────────────┘
                              │
                              │ WebSocket
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                    Supabase Realtime                        │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐           │
│  │  Postgres   │ │  Broadcast  │ │  Presence   │           │
│  │  Changes    │ │             │ │             │           │
│  └─────────────┘ └─────────────┘ └─────────────┘           │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                    PostgreSQL                               │
│              数据变更 → 实时推送                             │
└─────────────────────────────────────────────────────────────┘

Dashboard > Database > Replication

配置步骤
├── 1. 找到要启用实时订阅的表
├── 2. 开启Replication
└── 3. 选择要监听的列

-- 启用表的实时订阅
ALTER PUBLICATION supabase_realtime ADD TABLE messages;

-- 移除表的实时订阅
ALTER PUBLICATION supabase_realtime DROP TABLE messages;

-- 查看已启用的表
SELECT * FROM pg_publication_tables 
WHERE pubname = 'supabase_realtime';

// 创建频道
const channel = supabase.channel('my-channel')

// 订阅
channel.subscribe((status) => {
  if (status === 'SUBSCRIBED') {
    console.log('Connected!')
  }
})

// 监听所有变更
const channel = supabase
  .channel('schema-db-changes')
  .on(
    'postgres_changes',
    {
      event: '*',
      schema: 'public',
      table: 'messages',
    },
    (payload) => {
      console.log('Change received!', payload)
    }
  )
  .subscribe()

// 监听INSERT
const channel = supabase
  .channel('inserts')
  .on(
    'postgres_changes',
    {
      event: 'INSERT',
      schema: 'public',
      table: 'messages',
    },
    (payload) => {
      console.log('New message:', payload.new)
    }
  )
  .subscribe()

// 监听UPDATE
const channel = supabase
  .channel('updates')
  .on(
    'postgres_changes',
    {
      event: 'UPDATE',
      schema: 'public',
      table: 'messages',
    },
    (payload) => {
      console.log('Updated:', payload.old, '→', payload.new)
    }
  )
  .subscribe()

// 监听DELETE
const channel = supabase
  .channel('deletes')
  .on(
    'postgres_changes',
    {
      event: 'DELETE',
      schema: 'public',
      table: 'messages',
    },
    (payload) => {
      console.log('Deleted:', payload.old)
    }
  )
  .subscribe()

// 只监听特定用户的变更
const channel = supabase
  .channel('user-messages')
  .on(
    'postgres_changes',
    {
      event: '*',
      schema: 'public',
      table: 'messages',
      filter: 'user_id=eq.123',
    },
    (payload) => {
      console.log('User message changed:', payload)
    }
  )
  .subscribe()

// 等于
filter: 'column=eq.value'

// 不等于
filter: 'column=neq.value'

// 大于
filter: 'column=gt.value'

// 小于
filter: 'column=lt.value'

// 大于等于
filter: 'column=gte.value'

// 小于等于
filter: 'column=lte.value'

// IN
filter: 'column=in.(value1,value2,value3)'

// 取消订阅
await channel.unsubscribe()

// 取消所有频道
await supabase.removeAllChannels()

const channel = supabase
  .channel('status-channel')
  .subscribe((status) => {
    switch (status) {
      case 'SUBSCRIBED':
        console.log('Connected and subscribed')
        break
      case 'TIMED_OUT':
        console.log('Connection timed out')
        break
      case 'CLOSED':
        console.log('Connection closed')
        break
      case 'CHANNEL_ERROR':
        console.log('Channel error')
        break
    }
  })

const supabase = createClient(url, key, {
  realtime: {
    params: {
      eventsPerSecond: 10,  // 每秒事件数限制
    },
  },
})

实时订阅优化建议
├── 只订阅需要的表
├── 使用过滤减少数据量
├── 及时取消不用的订阅
├── 处理断线重连
└── 限制事件频率

const channel = supabase
  .channel('error-handling')
  .on('postgres_changes', { ... }, (payload) => {
    // 处理数据
  })
  .subscribe((status, err) => {
    if (status === 'CHANNEL_ERROR') {
      console.error('Subscription error:', err)
      // 重试或通知用户
    }
  })