websocket-realtime-builder

WebSocket 实时应用构建器

使用 WebSocket 和 Socket.io 构建实时应用。

核心工作流

1. 选择库：Socket.io 与原生 WebSocket 二选一
2. 设置服务端：配置 WebSocket 服务端
3. 添加身份验证：验证连接
4. 实现房间：对连接进行分组
5. 处理事件：定义事件处理程序
6. 添加重新连接：妥善处理断开连接

安装

# Server
npm install socket.io

# Client
npm install socket.io-client

服务端设置

基本 Socket.io 服务端

// server.ts
import express from 'express';
import { createServer } from 'http';
import { Server, Socket } from 'socket.io';
import { verifyToken } from './auth';

const app = express();
const httpServer = createServer(app);

const io = new Server(httpServer, {
  cors: {
    origin: process.env.CLIENT_URL,
    credentials: true,
  },
  pingInterval: 25000,
  pingTimeout: 60000,
});

// Authentication middleware
io.use(async (socket, next) => {
  const token = socket.handshake.auth.token;

  if (!token) {
    return next(new Error('Authentication required'));
  }

  try {
    const user = await verifyToken(token);
    socket.data.user = user;
    next();
  } catch (err) {
    next(new Error('Invalid token'));
  }
});

io.on('connection', (socket: Socket) => {
  const user = socket.data.user;
  console.log(`User connected: ${user.id}`);

  // Join user's personal room
  socket.join(`user:${user.id}`);

  // Handle events
  socket.on('disconnect', () => {
    console.log(`User disconnected: ${user.id}`);
  });
});

httpServer.listen(3001, () => {
  console.log('Socket.io server running on port 3001');
});

export { io };

命名空间和房间

// namespaces/chat.ts
import { Server, Socket } from 'socket.io';

export function setupChatNamespace(io: Server) {
  const chatNamespace = io.of('/chat');

  chatNamespace.on('connection', (socket: Socket) => {
    const user = socket.data.user;

    // Join a chat room
    socket.on('join-room', async (roomId: string) => {
      // Validate user can access this room
      const canAccess = await canAccessRoom(user.id, roomId);
      if (!canAccess) {
        socket.emit('error', { message: 'Access denied' });
        return;
      }

      socket.join(`room:${roomId}`);
      socket.to(`room:${roomId}`).emit('user-joined', {
        userId: user.id,
        name: user.name,
      });
    });

    // Leave a chat room
    socket.on('leave-room', (roomId: string) => {
      socket.leave(`room:${roomId}`);
      socket.to(`room:${roomId}`).emit('user-left', {
        userId: user.id,
      });
    });

    // Send message
    socket.on('send-message', async (data: { roomId: string; content: string }) => {
      const { roomId, content } = data;

      // Save to database
      const message = await db.message.create({
        data: {
          roomId,
          authorId: user.id,
          content,
        },
        include: { author: true },
      });

      // Broadcast to room
      chatNamespace.to(`room:${roomId}`).emit('new-message', {
        id: message.id,
        content: message.content,
        author: {
          id: user.id,
          name: user.name,
        },
        createdAt: message.createdAt,
      });
    });

    // Typing indicator
    socket.on('typing-start', (roomId: string) => {
      socket.to(`room:${roomId}`).emit('user-typing', {
        userId: user.id,
        name: user.name,
      });
    });

    socket.on('typing-stop', (roomId: string) => {
      socket.to(`room:${roomId}`).emit('user-stopped-typing', {
        userId: user.id,
      });
    });
  });

  return chatNamespace;
}

事件发射器

// services/notifications.ts
import { io } from '../server';

export class NotificationService {
  // Send to specific user
  static sendToUser(userId: string, event: string, data: any) {
    io.to(`user:${userId}`).emit(event, data);
  }

  // Send to multiple users
  static sendToUsers(userIds: string[], event: string, data: any) {
    userIds.forEach((userId) => {
      io.to(`user:${userId}`).emit(event, data);
    });
  }

  // Broadcast to all connected users
  static broadcast(event: string, data: any) {
    io.emit(event, data);
  }

  // Send to room
  static sendToRoom(roomId: string, event: string, data: any) {
    io.to(`room:${roomId}`).emit(event, data);
  }

  // Notify new order
  static notifyNewOrder(order: Order) {
    // Notify customer
    this.sendToUser(order.customerId, 'order:created', {
      orderId: order.id,
      status: order.status,
    });

    // Notify admins
    io.to('role:admin').emit('admin:new-order', {
      orderId: order.id,
      customer: order.customerName,
      total: order.total,
    });
  }
}

客户端设置

React 客户端 Hook

// hooks/useSocket.ts
import { useEffect, useRef, useState, useCallback } from 'react';
import { io, Socket } from 'socket.io-client';
import { useAuth } from './useAuth';

interface UseSocketOptions {
  namespace?: string;
  autoConnect?: boolean;
}

export function useSocket(options: UseSocketOptions = {}) {
  const { namespace = '/', autoConnect = true } = options;
  const { token } = useAuth();
  const socketRef = useRef<Socket | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  const [error, setError] = useState<string | null>(null);

  useEffect(() => {
    if (!token || !autoConnect) return;

    const socket = io(`${process.env.NEXT_PUBLIC_WS_URL}${namespace}`, {
      auth: { token },
      reconnection: true,
      reconnectionAttempts: 5,
      reconnectionDelay: 1000,
      reconnectionDelayMax: 5000,
    });

    socket.on('connect', () => {
      setIsConnected(true);
      setError(null);
    });

    socket.on('disconnect', () => {
      setIsConnected(false);
    });

    socket.on('connect_error', (err) => {
      setError(err.message);
      setIsConnected(false);
    });

    socketRef.current = socket;

    return () => {
      socket.disconnect();
    };
  }, [token, namespace, autoConnect]);

  const emit = useCallback((event: string, data?: any) => {
    socketRef.current?.emit(event, data);
  }, []);

  const on = useCallback((event: string, handler: (...args: any[]) => void) => {
    socketRef.current?.on(event, handler);
    return () => {
      socketRef.current?.off(event, handler);
    };
  }, []);

  const off = useCallback((event: string, handler?: (...args: any[]) => void) => {
    socketRef.current?.off(event, handler);
  }, []);

  return {
    socket: socketRef.current,
    isConnected,
    error,
    emit,
    on,
    off,
  };
}

聊天 Hook

// hooks/useChat.ts
import { useEffect, useState, useCallback } from 'react';
import { useSocket } from './useSocket';

interface Message {
  id: string;
  content: string;
  author: { id: string; name: string };
  createdAt: string;
}

interface TypingUser {
  userId: string;
  name: string;
}

export function useChat(roomId: string) {
  const { socket, isConnected, emit, on } = useSocket({ namespace: '/chat' });
  const [messages, setMessages] = useState<Message[]>([]);
  const [typingUsers, setTypingUsers] = useState<TypingUser[]>([]);

  // Join room on connect
  useEffect(() => {
    if (isConnected && roomId) {
      emit('join-room', roomId);

      return () => {
        emit('leave-room', roomId);
      };
    }
  }, [isConnected, roomId, emit]);

  // Listen for messages
  useEffect(() => {
    const unsubMessage = on('new-message', (message: Message) => {
      setMessages((prev) => [...prev, message]);
    });

    const unsubTyping = on('user-typing', (user: TypingUser) => {
      setTypingUsers((prev) => {
        if (prev.some((u) => u.userId === user.userId)) return prev;
        return [...prev, user];
      });
    });

    const unsubStopTyping = on('user-stopped-typing', ({ userId }: { userId: string }) => {
      setTypingUsers((prev) => prev.filter((u) => u.userId !== userId));
    });

    return () => {
      unsubMessage();
      unsubTyping();
      unsubStopTyping();
    };
  }, [on]);

  const sendMessage = useCallback((content: string) => {
    emit('send-message', { roomId, content });
  }, [emit, roomId]);

  const startTyping = useCallback(() => {
    emit('typing-start', roomId);
  }, [emit, roomId]);

  const stopTyping = useCallback(() => {
    emit('typing-stop', roomId);
  }, [emit, roomId]);

  return {
    messages,
    typingUsers,
    sendMessage,
    startTyping,
    stopTyping,
    isConnected,
  };
}

聊天组件

// components/ChatRoom.tsx
'use client';

import { useState, useRef, useEffect } from 'react';
import { useChat } from '@/hooks/useChat';

export function ChatRoom({ roomId }: { roomId: string }) {
  const {
    messages,
    typingUsers,
    sendMessage,
    startTyping,
    stopTyping,
    isConnected,
  } = useChat(roomId);

  const [input, setInput] = useState('');
  const messagesEndRef = useRef<HTMLDivElement>(null);
  const typingTimeoutRef = useRef<NodeJS.Timeout>();

  // Auto-scroll to bottom
  useEffect(() => {
    messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
  }, [messages]);

  const handleInputChange = (e: React.ChangeEvent<HTMLInputElement>) => {
    setInput(e.target.value);

    // Debounced typing indicator
    startTyping();
    clearTimeout(typingTimeoutRef.current);
    typingTimeoutRef.current = setTimeout(stopTyping, 2000);
  };

  const handleSubmit = (e: React.FormEvent) => {
    e.preventDefault();
    if (!input.trim()) return;

    sendMessage(input);
    setInput('');
    stopTyping();
  };

  if (!isConnected) {
    return <div>Connecting...</div>;
  }

  return (
    <div className="flex flex-col h-full">
      <div className="flex-1 overflow-y-auto p-4 space-y-4">
        {messages.map((message) => (
          <div key={message.id} className="flex gap-2">
            <span className="font-semibold">{message.author.name}:</span>
            <span>{message.content}</span>
          </div>
        ))}
        <div ref={messagesEndRef} />
      </div>

      {typingUsers.length > 0 && (
        <div className="px-4 py-2 text-sm text-gray-500">
          {typingUsers.map((u) => u.name).join(', ')}{' '}
          {typingUsers.length === 1 ? 'is' : 'are'} typing...
        </div>
      )}

      <form onSubmit={handleSubmit} className="p-4 border-t">
        <input
          type="text"
          value={input}
          onChange={handleInputChange}
          placeholder="Type a message..."
          className="w-full px-4 py-2 border rounded"
        />
      </form>
    </div>
  );
}

实时通知

// hooks/useNotifications.ts
import { useEffect, useState } from 'react';
import { useSocket } from './useSocket';
import { toast } from 'sonner';

interface Notification {
  id: string;
  type: string;
  title: string;
  message: string;
  createdAt: string;
  read: boolean;
}

export function useNotifications() {
  const { on, isConnected } = useSocket();
  const [notifications, setNotifications] = useState<Notification[]>([]);
  const [unreadCount, setUnreadCount] = useState(0);

  useEffect(() => {
    const unsubscribe = on('notification', (notification: Notification) => {
      setNotifications((prev) => [notification, ...prev]);
      setUnreadCount((prev) => prev + 1);

      // Show toast
      toast(notification.title, {
        description: notification.message,
      });
    });

    return unsubscribe;
  }, [on]);

  const markAsRead = (id: string) => {
    setNotifications((prev) =>
      prev.map((n) => (n.id === id ? { ...n, read: true } : n))
    );
    setUnreadCount((prev) => Math.max(0, prev - 1));
  };

  const markAllAsRead = () => {
    setNotifications((prev) => prev.map((n) => ({ ...n, read: true })));
    setUnreadCount(0);
  };

  return {
    notifications,
    unreadCount,
    markAsRead,
    markAllAsRead,
    isConnected,
  };
}

在线状态系统

// server/presence.ts
import { Server, Socket } from 'socket.io';

interface UserPresence {
  odlineserId: string;
  name: string;
  status: 'online' | 'away' | 'busy';
  lastSeen: Date;
}

const onlineUsers = new Map<string, UserPresence>();

export function setupPresence(io: Server) {
  io.on('connection', (socket: Socket) => {
    const user = socket.data.user;

    // User comes online
    onlineUsers.set(user.id, {
      odlineserId: user.id,
      name: user.name,
      status: 'online',
      lastSeen: new Date(),
    });

    // Broadcast to all users
    io.emit('presence:update', {
      userId: user.id,
      status: 'online',
    });

    // Send current online users to new connection
    socket.emit('presence:list', Array.from(onlineUsers.values()));

    // Handle status changes
    socket.on('presence:status', (status: 'online' | 'away' | 'busy') => {
      const presence = onlineUsers.get(user.id);
      if (presence) {
        presence.status = status;
        io.emit('presence:update', { userId: user.id, status });
      }
    });

    // Handle disconnect
    socket.on('disconnect', () => {
      onlineUsers.delete(user.id);
      io.emit('presence:update', {
        userId: user.id,
        status: 'offline',
      });
    });
  });
}

错误处理与重新连接

// Client-side error handling
const socket = io(WS_URL, {
  auth: { token },
  reconnection: true,
  reconnectionAttempts: 10,
  reconnectionDelay: 1000,
  reconnectionDelayMax: 10000,
  timeout: 20000,
});

socket.on('connect_error', (error) => {
  if (error.message === 'Authentication required') {
    // Redirect to login
    router.push('/login');
  } else {
    console.error('Connection error:', error);
  }
});

socket.on('reconnect', (attemptNumber) => {
  console.log(`Reconnected after ${attemptNumber} attempts`);
  // Re-sync state
  socket.emit('sync-state');
});

socket.on('reconnect_failed', () => {
  console.error('Failed to reconnect');
  // Show user notification
  toast.error('Connection lost. Please refresh the page.');
});

最佳实践

1. 验证连接：连接时验证令牌
2. 使用房间：按逻辑对用户进行分组
3. 处理重新连接：重新连接后重新同步状态
4. 对事件进行防抖：防止事件泛滥（例如，输入提示）
5. 清理监听器：组件卸载时移除
6. 确认重要事件：确认关键消息
7. 使用命名空间：分离不同关注点（聊天、通知）
8. 使用 Redis 扩展：在多台服务端上使用 Redis 适配器

输出检查清单

每个 WebSocket 实现都应包含：

• 带身份验证中间件的服务端
• 带重新连接处理的客户端 Hook
• 房间／命名空间组织
• 事件类型定义
• 错误处理与恢复
• 聊天输入提示
• 在线状态系统（在线／离线）
• 正确处理断开连接
• 事件速率限制
• 用于扩展的 Redis 适配器（生产环境）