Go 및 WebSocket을 사용하여 실시간 협업 플랫폼 구축

소개

여러 사용자가 동시에 함께 작업할 수 있는 분산형 실시간 협업 플랫폼을 구축해 보겠습니다. 이 프로젝트에서는 Go의 WebSocket 처리, 충돌 해결 및 상태 동기화를 시연합니다.

프로젝트 개요: 실시간 협업 플랫폼

핵심 기능

• 실시간 문서 편집
• 커서 위치 동기화
• 존재 인식
• 운영 혁신
• 갈등 해결
• 채팅 기능

기술적 구현

1. 웹소켓 서버

// WebSocket server implementation
type CollaborationServer struct {
    sessions    map[string]*Session
    documents   map[string]*Document
    broadcast   chan Message
    register    chan *Client
    unregister  chan *Client
}

type Client struct {
    id       string
    session  *Session
    conn     *websocket.Conn
    send     chan Message
}

type Message struct {
    Type    MessageType `json:"type"`
    Payload interface{} `json:"payload"`
}

func NewCollaborationServer() *CollaborationServer {
    return &CollaborationServer{
        sessions:   make(map[string]*Session),
        documents:  make(map[string]*Document),
        broadcast:  make(chan Message),
        register:   make(chan *Client),
        unregister: make(chan *Client),
    }
}

func (s *CollaborationServer) Run() {
    for {
        select {
        case client := <-s.register:
            s.handleRegister(client)

        case client := <-s.unregister:
            s.handleUnregister(client)

        case message := <-s.broadcast:
            s.handleBroadcast(message)
        }
    }
}

func (s *CollaborationServer) handleRegister(client *Client) {
    session := s.sessions[client.session.ID]
    if session == nil {
        session = &Session{
            ID:      client.session.ID,
            Clients: make(map[string]*Client),
        }
        s.sessions[session.ID] = session
    }
    session.Clients[client.id] = client
}

2. 운영 혁신 엔진

// Operational transformation implementation
type Operation struct {
    Type      OperationType
    Position  int
    Content   string
    ClientID  string
    Revision  int
}

type Document struct {
    ID        string
    Content   string
    History   []Operation
    Revision  int
    mu        sync.RWMutex
}

func (d *Document) ApplyOperation(op Operation) error {
    d.mu.Lock()
    defer d.mu.Unlock()

    // Transform operation against concurrent operations
    transformedOp := d.transformOperation(op)

    // Apply the transformed operation
    switch transformedOp.Type {
    case OpInsert:
        d.insertContent(transformedOp.Position, transformedOp.Content)
    case OpDelete:
        d.deleteContent(transformedOp.Position, len(transformedOp.Content))
    }

    // Update revision and history
    d.Revision++
    d.History = append(d.History, transformedOp)

    return nil
}

func (d *Document) transformOperation(op Operation) Operation {
    transformed := op

    // Transform against all concurrent operations
    for _, historical := range d.History[op.Revision:] {
        transformed = transform(transformed, historical)
    }

    return transformed
}

3. 출석 시스템

// Real-time presence tracking
type PresenceSystem struct {
    mu       sync.RWMutex
    users    map[string]*UserPresence
    updates  chan PresenceUpdate
}

type UserPresence struct {
    UserID    string
    Document  string
    Cursor    Position
    Selection Selection
    LastSeen  time.Time
}

type Position struct {
    Line   int
    Column int
}

type Selection struct {
    Start Position
    End   Position
}

func (ps *PresenceSystem) UpdatePresence(update PresenceUpdate) {
    ps.mu.Lock()
    defer ps.mu.Unlock()

    user := ps.users[update.UserID]
    if user == nil {
        user = &UserPresence{UserID: update.UserID}
        ps.users[update.UserID] = user
    }

    user.Document = update.Document
    user.Cursor = update.Cursor
    user.Selection = update.Selection
    user.LastSeen = time.Now()

    // Broadcast update to other users
    ps.updates <- update
}

func (ps *PresenceSystem) StartCleanup() {
    ticker := time.NewTicker(30 * time.Second)
    go func() {
        for range ticker.C {
            ps.cleanupInactiveUsers()
        }
    }()
}

4. 갈등 해결

// Conflict resolution system
type ConflictResolver struct {
    strategy ConflictStrategy
}

type ConflictStrategy interface {
    Resolve(a, b Operation) Operation
}

// Last-write-wins strategy
type LastWriteWinsStrategy struct{}

func (s *LastWriteWinsStrategy) Resolve(a, b Operation) Operation {
    if a.Timestamp.After(b.Timestamp) {
        return a
    }
    return b
}

// Three-way merge strategy
type ThreeWayMergeStrategy struct{}

func (s *ThreeWayMergeStrategy) Resolve(base, a, b Operation) Operation {
    // Implement three-way merge logic
    if a.Position == b.Position {
        if a.Type == OpDelete && b.Type == OpDelete {
            return a // Both deleted same content
        }
        if a.Timestamp.After(b.Timestamp) {
            return a
        }
        return b
    }

    // Non-overlapping changes
    if a.Position < b.Position {
        return combineOperations(a, b)
    }
    return combineOperations(b, a)
}

5. 상태 동기화

// State synchronization system
type SyncManager struct {
    documents map[string]*DocumentState
    clients   map[string]*ClientState
}

type DocumentState struct {
    Content    string
    Version    int64
    Operations []Operation
    Checksum   string
}

type ClientState struct {
    LastSync    time.Time
    SyncVersion int64
}

func (sm *SyncManager) SynchronizeState(clientID string, docID string) error {
    client := sm.clients[clientID]
    doc := sm.documents[docID]

    if client.SyncVersion == doc.Version {
        return nil // Already in sync
    }

    // Get operations since last sync
    ops := sm.getOperationsSince(docID, client.SyncVersion)

    // Apply operations to client state
    for _, op := range ops {
        if err := sm.applyOperation(clientID, op); err != nil {
            return fmt.Errorf("sync failed: %w", err)
        }
    }

    // Update client sync version
    client.SyncVersion = doc.Version
    client.LastSync = time.Now()

    return nil
}

6. 채팅 시스템

// Real-time chat implementation
type ChatSystem struct {
    rooms    map[string]*ChatRoom
    history  map[string][]ChatMessage
}

type ChatRoom struct {
    ID        string
    Members   map[string]*Client
    Messages  chan ChatMessage
}

type ChatMessage struct {
    ID        string
    RoomID    string
    UserID    string
    Content   string
    Timestamp time.Time
}

func (cs *ChatSystem) SendMessage(msg ChatMessage) error {
    room := cs.rooms[msg.RoomID]
    if room == nil {
        return fmt.Errorf("room not found: %s", msg.RoomID)
    }

    // Store message in history
    cs.history[msg.RoomID] = append(cs.history[msg.RoomID], msg)

    // Broadcast to room members
    room.Messages <- msg

    return nil
}

고급 기능

1. 성능 최적화

• 메시지 일괄 처리
• 연산 압축
• 선택방송

// Message batching implementation
type MessageBatcher struct {
    messages []Message
    timeout  time.Duration
    size     int
    batch    chan []Message
}

func (mb *MessageBatcher) Add(msg Message) {
    mb.messages = append(mb.messages, msg)

    if len(mb.messages) >= mb.size {
        mb.flush()
    }
}

func (mb *MessageBatcher) Start() {
    ticker := time.NewTicker(mb.timeout)
    go func() {
        for range ticker.C {
            mb.flush()
        }
    }()
}

2. 확장 고려사항

// Distributed coordination using Redis
type DistributedCoordinator struct {
    client  *redis.Client
    pubsub  *redis.PubSub
}

func (dc *DistributedCoordinator) PublishUpdate(update Update) error {
    return dc.client.Publish(ctx, "updates", update).Err()
}

func (dc *DistributedCoordinator) SubscribeToUpdates() {
    sub := dc.client.Subscribe(ctx, "updates")
    for msg := range sub.Channel() {
        // Handle distributed update
        dc.handleUpdate(msg)
    }
}

테스트 전략

1. 단위 테스트

func TestOperationalTransformation(t *testing.T) {
    doc := NewDocument("test")

    // Test concurrent inserts
    op1 := Operation{Type: OpInsert, Position: 0, Content: "Hello"}
    op2 := Operation{Type: OpInsert, Position: 0, Content: "World"}

    doc.ApplyOperation(op1)
    doc.ApplyOperation(op2)

    expected := "WorldHello"
    if doc.Content != expected {
        t.Errorf("expected %s, got %s", expected, doc.Content)
    }
}

2. 통합 테스트

func TestRealTimeCollaboration(t *testing.T) {
    server := NewCollaborationServer()
    go server.Run()

    // Create test clients
    client1 := createTestClient()
    client2 := createTestClient()

    // Simulate concurrent editing
    go simulateEditing(client1)
    go simulateEditing(client2)

    // Verify final state
    time.Sleep(2 * time.Second)
    verifyDocumentState(t, server)
}

배포 아키텍처

• 로드 밸런서 뒤에 있는 여러 서버 인스턴스
• pub/sub 및 주 조정을 위한 Redis
• WebSocket 연결 관리
• 모니터링 및 알림

결론

실시간 협업 플랫폼 구축은 복잡한 분산 시스템 개념과 실시간 데이터 동기화를 보여줍니다. 이 프로젝트는 Go의 강력한 동시성 기능과 WebSocket 처리 기능을 보여줍니다.

추가 리소스

• 웹소켓 프로토콜 RFC
• 운영 혁신
• Redis Pub/Sub 문서

실시간 협업 시스템을 구축한 경험을 댓글로 공유해주세요!

---

태그: #golang #websockets #realtime #collaboration #distributed-systems

위 내용은 Go 및 WebSocket을 사용하여 실시간 협업 플랫폼 구축의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!