4.8.2 拓展：数据一变就通知我——实时订阅：数据变更推送

一句话破题

Supabase Realtime 让你的应用能够"监听"数据库变化——数据一变，客户端立即知道，无需轮询。

Realtime 工作原理

启用 Realtime

在 Dashboard 中启用：

1. 进入 Table Editor
2. 选择要监听的表
3. 启用 Realtime

使用 SQL 启用：

-- 启用表的实时功能
ALTER PUBLICATION supabase_realtime ADD TABLE posts;

-- 验证
SELECT * FROM pg_publication_tables WHERE pubname = 'supabase_realtime';

订阅表变更

import { supabase } from '@/lib/supabase'

// 订阅所有变更
const channel = supabase
  .channel('posts-changes')
  .on(
    'postgres_changes',
    {
      event: '*',  // 'INSERT' | 'UPDATE' | 'DELETE' | '*'
      schema: 'public',
      table: 'posts'
    },
    (payload) => {
      console.log('Change received!', payload)
    }
  )
  .subscribe()

// 取消订阅
channel.unsubscribe()

监听特定事件

// 只监听新插入的数据
const channel = supabase
  .channel('new-posts')
  .on(
    'postgres_changes',
    {
      event: 'INSERT',
      schema: 'public',
      table: 'posts'
    },
    (payload) => {
      console.log('New post:', payload.new)
    }
  )
  .subscribe()

过滤订阅

// 只监听特定用户的文章
const channel = supabase
  .channel('user-posts')
  .on(
    'postgres_changes',
    {
      event: '*',
      schema: 'public',
      table: 'posts',
      filter: `author_id=eq.${userId}`
    },
    (payload) => {
      console.log('User post changed:', payload)
    }
  )
  .subscribe()

React Hook 封装

'use client'

import { useEffect, useState } from 'react'
import { supabase } from '@/lib/supabase'
import { RealtimePostgresChangesPayload } from '@supabase/supabase-js'

interface Post {
  id: string
  title: string
  content: string
}

export function usePosts() {
  const [posts, setPosts] = useState<Post[]>([])
  
  useEffect(() => {
    // 初始加载
    async function loadPosts() {
      const { data } = await supabase
        .from('posts')
        .select('*')
        .order('created_at', { ascending: false })
      
      setPosts(data ?? [])
    }
    
    loadPosts()
    
    // 订阅变更
    const channel = supabase
      .channel('posts-realtime')
      .on(
        'postgres_changes',
        { event: '*', schema: 'public', table: 'posts' },
        (payload: RealtimePostgresChangesPayload<Post>) => {
          if (payload.eventType === 'INSERT') {
            setPosts(prev => [payload.new, ...prev])
          } else if (payload.eventType === 'UPDATE') {
            setPosts(prev => 
              prev.map(p => p.id === payload.new.id ? payload.new : p)
            )
          } else if (payload.eventType === 'DELETE') {
            setPosts(prev => 
              prev.filter(p => p.id !== payload.old.id)
            )
          }
        }
      )
      .subscribe()
    
    return () => {
      channel.unsubscribe()
    }
  }, [])
  
  return posts
}

Presence：在线状态

// 追踪用户在线状态
const channel = supabase.channel('online-users')

// 追踪当前用户
channel.on('presence', { event: 'sync' }, () => {
  const state = channel.presenceState()
  console.log('在线用户:', Object.keys(state))
})

channel.on('presence', { event: 'join' }, ({ key, newPresences }) => {
  console.log('用户上线:', key, newPresences)
})

channel.on('presence', { event: 'leave' }, ({ key, leftPresences }) => {
  console.log('用户离线:', key, leftPresences)
})

// 订阅并加入
channel.subscribe(async (status) => {
  if (status === 'SUBSCRIBED') {
    await channel.track({
      user_id: userId,
      online_at: new Date().toISOString()
    })
  }
})

Broadcast：广播消息

// 创建聊天室
const channel = supabase.channel('chat-room')

// 监听消息
channel.on('broadcast', { event: 'message' }, ({ payload }) => {
  console.log('收到消息:', payload)
})

channel.subscribe()

// 发送消息
channel.send({
  type: 'broadcast',
  event: 'message',
  payload: {
    user: 'Alice',
    text: 'Hello!'
  }
})

聊天室完整示例

'use client'

import { useEffect, useState, useRef } from 'react'
import { supabase } from '@/lib/supabase'

interface Message {
  id: string
  user: string
  text: string
  timestamp: string
}

export function ChatRoom({ userId }: { userId: string }) {
  const [messages, setMessages] = useState<Message[]>([])
  const [input, setInput] = useState('')
  const channelRef = useRef<ReturnType<typeof supabase.channel>>()
  
  useEffect(() => {
    const channel = supabase.channel('chat-room')
    
    channel.on('broadcast', { event: 'message' }, ({ payload }) => {
      setMessages(prev => [...prev, payload as Message])
    })
    
    channel.subscribe()
    channelRef.current = channel
    
    return () => {
      channel.unsubscribe()
    }
  }, [])
  
  async function sendMessage() {
    if (!input.trim() || !channelRef.current) return
    
    const message: Message = {
      id: crypto.randomUUID(),
      user: userId,
      text: input,
      timestamp: new Date().toISOString()
    }
    
    await channelRef.current.send({
      type: 'broadcast',
      event: 'message',
      payload: message
    })
    
    setInput('')
  }
  
  return (
    <div>
      <div className="messages">
        {messages.map(msg => (
          <div key={msg.id}>
            <strong>{msg.user}:</strong> {msg.text}
          </div>
        ))}
      </div>
      <input
        value={input}
        onChange={e => setInput(e.target.value)}
        onKeyDown={e => e.key === 'Enter' && sendMessage()}
      />
      <button onClick={sendMessage}>发送</button>
    </div>
  )
}

本节小结

• postgres_changes 监听数据库表的变更
• presence 追踪用户在线状态
• broadcast 实现广播消息
• 记得在组件卸载时取消订阅