Какую архитектуру лучше использовать для большого форума?

Архитектура для большого форума

Требования большого форума

Перед выбором архитектуры нужно понять требования:

• Миллионы пользователей
• Высокая нагрузка на чтение (просмотры постов)
• Модеративные нагрузки на запись (создание постов)
• Сложные запросы (фильтры, поиск, рейтинги)
• Низкая задержка (latency < 100ms)

Рекомендуемая архитектура

┌─────────────────────────────────────────────────────┐
│              CDN (CloudFlare, Cloudfront)           │
│  Кэширование статики и API ответов                 │
└────────────────┬────────────────────────────────────┘
                 │
┌────────────────▼────────────────────────────────────┐
│          Load Balancer (Nginx, HAProxy)             │
│  Распределение трафика, SSL/TLS termination        │
└────────────────┬────────────────────────────────────┘
                 │
      ┌──────────┴──────────┐
      │                     │
┌─────▼──────┐     ┌──────▼────────┐
│ API Node 1 │     │ API Node 2    │
│ (FastAPI)  │     │ (FastAPI)     │
└─────┬──────┘     └──────┬────────┘
      │                   │
┌─────▼───────────────────▼──────────┐
│     Main Database (PostgreSQL)     │
│  - Таблицы форума                 │
│  - Пользователи, посты, комменты  │
│  - Рейтинги, подписки             │
└────────────────┬────────────────────┘
                 │
     ┌───────────┼───────────┐
     │           │           │
  ┌──▼───┐  ┌───▼──┐  ┌─────▼──┐
  │Redis │  │Search│  │Archive │
  │Cache │  │Engine│  │Storage │
  │      │  │(ES)  │  │(S3)    │
  └──────┘  └──────┘  └────────┘

1. Backend (Python + FastAPI)

Структура проекта

forum-api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── dependencies.py
│   ├── database.py
│   ├── cache.py
│   ├── schemas/  # Pydantic модели
│   │   ├── user.py
│   │   ├── post.py
│   │   ├── comment.py
│   │   └── pagination.py
│   ├── models/  # SQLAlchemy модели
│   │   ├── user.py
│   │   ├── post.py
│   │   ├── comment.py
│   │   └── rating.py
│   ├── api/
│   │   ├── v1/
│   │   │   ├── posts.py
│   │   │   ├── comments.py
│   │   │   ├── users.py
│   │   │   └── search.py
│   │   └── router.py
│   ├── services/  # Бизнес логика
│   │   ├── post_service.py
│   │   ├── comment_service.py
│   │   ├── user_service.py
│   │   └── notification_service.py
│   ├── repositories/  # Доступ к данным
│   │   ├── post_repo.py
│   │   ├── comment_repo.py
│   │   └── user_repo.py
│   └── utils/
│       ├── cache.py
│       ├── pagination.py
│       └── validators.py
├── tests/
├── requirements.txt
└── main.py

Основной файл main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.compression import GZipMiddleware
from contextlib import asynccontextmanager

from app.config import settings
from app.database import engine, Base
from app.api.router import router
from app.cache import redis_client

# Инициализация БД
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    Base.metadata.create_all(bind=engine)
    await redis_client.connect()
    yield
    # Shutdown
    await redis_client.close()

app = FastAPI(
    title="Forum API",
    version="1.0.0",
    lifespan=lifespan
)

# Middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(GZipMiddleware, minimum_size=1000)  # Сжатие ответов

# Роутеры
app.include_router(router)

@app.get("/health")
async def health_check():
    return {"status": "ok"}

2. База данных

Основные таблицы

from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey, Boolean, Index
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    username = Column(String(100), unique=True, index=True)
    email = Column(String(255), unique=True, index=True)
    password_hash = Column(String(255))
    created_at = Column(DateTime, default=datetime.utcnow, index=True)
    reputation = Column(Integer, default=0)  # Для быстрой фильтрации
    
    __table_args__ = (
        Index("idx_user_created", "created_at"),
        Index("idx_user_reputation", "reputation"),
    )

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"), index=True)
    title = Column(String(300), index=True)
    content = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow, index=True)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    views_count = Column(Integer, default=0)  # Денормализация для производительности
    rating = Column(Integer, default=0, index=True)
    is_deleted = Column(Boolean, default=False, index=True)
    
    __table_args__ = (
        Index("idx_post_created", "created_at"),
        Index("idx_post_rating", "rating"),
        Index("idx_post_user_created", "user_id", "created_at"),
    )

class Comment(Base):
    __tablename__ = "comments"
    
    id = Column(Integer, primary_key=True)
    post_id = Column(Integer, ForeignKey("posts.id"), index=True)
    user_id = Column(Integer, ForeignKey("users.id"), index=True)
    content = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow, index=True)
    rating = Column(Integer, default=0, index=True)
    is_deleted = Column(Boolean, default=False)
    
    __table_args__ = (
        Index("idx_comment_post", "post_id", "created_at"),
        Index("idx_comment_user", "user_id", "created_at"),
    )

3. Кэширование

import redis.asyncio as redis
import json

class CacheManager:
    def __init__(self, redis_url: str):
        self.redis = None
        self.redis_url = redis_url
    
    async def connect(self):
        self.redis = await redis.from_url(self.redis_url)
    
    async def close(self):
        if self.redis:
            await self.redis.close()
    
    async def get_post(self, post_id: int):
        """Получить пост из кэша"""
        cached = await self.redis.get(f"post:{post_id}")
        return json.loads(cached) if cached else None
    
    async def set_post(self, post_id: int, data: dict, ttl: int = 3600):
        """Кэшировать пост на 1 час"""
        await self.redis.setex(
            f"post:{post_id}",
            ttl,
            json.dumps(data)
        )
    
    async def invalidate_post(self, post_id: int):
        """Инвалидировать кэш поста при обновлении"""
        await self.redis.delete(f"post:{post_id}")
    
    async def get_feed(self, user_id: int, page: int):
        """Кэшированная лента"""
        cached = await self.redis.get(f"feed:{user_id}:page:{page}")
        return json.loads(cached) if cached else None
    
    async def set_feed(self, user_id: int, page: int, data: list, ttl: int = 600):
        """Кэшировать ленту на 10 минут"""
        await self.redis.setex(
            f"feed:{user_id}:page:{page}",
            ttl,
            json.dumps(data)
        )

cache = CacheManager(redis_url="redis://localhost:6379/0")

4. API Endpoints

from fastapi import APIRouter, Query, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

from app.schemas.post import PostResponse, PostCreate, PostUpdate
from app.models.post import Post
from app.services.post_service import PostService

router = APIRouter(prefix="/api/v1", tags=["posts"])
post_service = PostService()

@router.get("/posts")
async def get_posts(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    sort_by: str = Query("created_at", enum=["created_at", "rating", "views"]),
    cache = Depends(lambda: cache)
) -> List[PostResponse]:
    """Получить список постов с пагинацией"""
    # Проверяем кэш
    cached = await cache.get_feed(user_id=0, page=skip // limit)
    if cached:
        return cached
    
    # Получаем из БД
    posts = await post_service.get_posts(
        skip=skip,
        limit=limit,
        sort_by=sort_by
    )
    
    # Кэшируем
    await cache.set_feed(user_id=0, page=skip // limit, data=posts)
    
    return posts

@router.get("/posts/{post_id}")
async def get_post(post_id: int) -> PostResponse:
    """Получить конкретный пост"""
    # Проверяем кэш
    cached = await cache.get_post(post_id)
    if cached:
        return cached
    
    # Получаем из БД
    post = await post_service.get_post(post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    
    # Кэшируем
    await cache.set_post(post_id, post.dict())
    
    # Увеличиваем счетчик просмотров в background
    await post_service.increment_views(post_id)
    
    return post

@router.post("/posts")
async def create_post(post: PostCreate) -> PostResponse:
    """Создать новый пост"""
    new_post = await post_service.create_post(post)
    return new_post

@router.put("/posts/{post_id}")
async def update_post(post_id: int, post: PostUpdate) -> PostResponse:
    """Обновить пост"""
    updated = await post_service.update_post(post_id, post)
    
    # Инвалидировать кэш
    await cache.invalidate_post(post_id)
    
    return updated

5. Поиск (Elasticsearch)

from elasticsearch import Elasticsearch

class SearchService:
    def __init__(self, es_host: str):
        self.es = Elasticsearch([es_host])
    
    def index_post(self, post_id: int, title: str, content: str):
        """Добавить пост в индекс поиска"""
        self.es.index(
            index="posts",
            id=post_id,
            body={
                "title": title,
                "content": content,
                "created_at": datetime.utcnow().isoformat()
            }
        )
    
    def search(self, query: str, limit: int = 20):
        """Полнотекстовый поиск по постам"""
        result = self.es.search(
            index="posts",
            body={
                "query": {
                    "multi_match": {
                        "query": query,
                        "fields": ["title^2", "content"]
                    }
                },
                "size": limit
            }
        )
        return [hit["_source"] for hit in result["hits"]["hits"]]

search_service = SearchService(es_host="localhost:9200")

Ключевые стратегии для масштабирования

1. Database Indexing — индексы на часто используемые поля
2. Caching Layer — Redis для горячих данных
3. Read Replicas — несколько реплик БД для чтения
4. Async/Await — неблокирующие операции
5. Connection Pooling — переиспользование соединений
6. Search Engine — Elasticsearch для полнотекстового поиска
7. Message Queue — RabbitMQ для асинхронных задач (уведомления, индексация)
8. CDN — кэширование статики
9. Database Denormalization — denormalize fields for speed
10. Monitoring & Logging — для отладки проблем масштабирования

Этот подход позволит форуму комфортно обслуживать миллионы пользователей.