首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >从单文件到微服务雏形:Python 项目多应用架构的 3 种演进方案(附 FastAPI 实战代码)

从单文件到微服务雏形:Python 项目多应用架构的 3 种演进方案(附 FastAPI 实战代码)

作者头像
玄同765
发布2026-01-14 14:24:55
发布2026-01-14 14:24:55
390
举报
在这里插入图片描述
在这里插入图片描述

【个人主页:】

大语言模型(LLM)开发工程师中国传媒大学·数字媒体技术(智能交互与游戏设计) 深耕领域:大语言模型开发 / RAG知识库 / AI Agent落地 / 模型微调 技术栈:Python / LangChain/RAG(Dify+Redis+Milvus)| SQL/NumPy | FastAPI+Docker ️ 工程能力:专注模型工程化部署、知识库构建与优化,擅长全流程解决方案 专栏传送门:LLM大模型开发 项目实战指南Python 从真零基础到纯文本 LLM 全栈实战​​​​​从零学 SQL + 大模型应用落地大模型开发小白专属:从 0 入门 Linux&Shell 「让AI交互更智能,让技术落地更高效」 欢迎技术探讨/项目合作! 关注我,解锁大模型与智能交互的无限可能!

摘要

你是不是写 Python 项目时,把所有 API 接口、业务逻辑、工具函数都堆在一个 app 目录下,上线后改登录模块要动整个后端,测试时要跑所有模块?这篇讲单文件→扁平多应用→分层多应用→微服务雏形的 3 种演进方案,附 FastAPI 的学生管理系统 + 课程管理系统 + 共享组件的实战代码。


一、引言:为什么要做多应用结构?

在 Python 项目的开发初期,很多开发者会选择单文件或单 app 目录的结构,这种结构简单易上手,适合快速验证想法。但随着项目的复杂度提高,这种结构会导致以下问题:

  1. 代码可读性差:所有代码堆积在一个文件或一个目录下,超过 1000 行就难以阅读和维护;
  2. 耦合度高:修改一个模块时可能会影响其他模块,如修改用户管理模块的密码加密函数,可能会影响课程管理模块的注册功能;
  3. 测试困难:所有功能都在一个应用中,测试时需要跑所有模块,测试覆盖率难以提高;
  4. 部署复杂:无法独立部署某个模块,如只修改了支付模块,需要重新部署整个项目;
  5. 团队协作困难:多个开发者同时开发同一个应用,容易出现代码冲突。

多应用结构的出现解决了这些问题,它允许我们将项目的功能模块拆分为多个独立的应用,每个应用负责一个特定的业务功能,提高代码的可读性、可维护性、可测试性和可扩展性。


二、应用结构演进路径
2.1 阶段一:单文件(入门级,代码堆积)

这是 Python 项目的最简单结构,所有代码都堆在一个 main.py 文件中。

代码示例(学生管理系统)

代码语言:javascript
复制
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List

app = FastAPI(title="学生管理系统", version="1.0.0")

# 数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 密码加密
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# JWT配置
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# OAuth2配置
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

# 数据库依赖
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 数据模型
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(20), unique=True, index=True, nullable=False)
    email = Column(String(50), unique=True, index=True, nullable=False)
    hashed_password = Column(String(100), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default="now()")
    updated_at = Column(DateTime(timezone=True), onupdate="now()")

class Student(Base):
    __tablename__ = "students"
    id = Column(Integer, primary_key=True, index=True)
    student_no = Column(String(20), unique=True, index=True, nullable=False)
    name = Column(String(50), nullable=False)
    gender = Column(String(10), nullable=False)
    birthdate = Column(String(10), nullable=True)
    phone = Column(String(20), nullable=True)
    email = Column(String(50), nullable=True)
    address = Column(String(200), nullable=True)
    user_id = Column(Integer, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default="now()")
    updated_at = Column(DateTime(timezone=True), onupdate="now()")

Base.metadata.create_all(bind=engine)

# 数据验证模型
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=20, pattern=r"^[a-zA-Z0-9_]+$")
    email: EmailStr
    password: str = Field(..., min_length=8, max_length=20, pattern=r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d).*$")

class UserInfo(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool = True
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    class Config:
        orm_mode = True

class StudentCreate(BaseModel):
    student_no: str = Field(..., min_length=8, max_length=20, pattern=r"^[a-zA-Z0-9]+$")
    name: str = Field(..., min_length=2, max_length=50)
    gender: str = Field(..., pattern=r"^男|女$")
    birthdate: Optional[str] = Field(None, pattern=r"^\d{4}-\d{2}-\d{2}$")
    phone: Optional[str] = Field(None, pattern=r"^1[3-9]\d{9}$")
    email: Optional[EmailStr] = None
    address: Optional[str] = Field(None, max_length=200)

class StudentInfo(BaseModel):
    id: int
    student_no: str
    name: str
    gender: str
    birthdate: Optional[str] = None
    phone: Optional[str] = None
    email: Optional[EmailStr] = None
    address: Optional[str] = None
    user_id: int
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    class Config:
        orm_mode = True

# 工具函数
def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def decode_access_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            return None
        return {"username": username}
    except JWTError:
        return None

# 业务逻辑
def get_user_by_username(db: Session, username: str):
    return db.query(User).filter(User.username == username).first()

def create_user(db: Session, user: UserCreate):
    hashed_password = get_password_hash(user.password)
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def authenticate_user(db: Session, username: str, password: str):
    user = get_user_by_username(db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

def get_student_by_no(db: Session, student_no: str):
    return db.query(Student).filter(Student.student_no == student_no).first()

def create_student(db: Session, student: StudentCreate, user_id: int):
    db_student = Student(
        student_no=student.student_no,
        name=student.name,
        gender=student.gender,
        birthdate=student.birthdate,
        phone=student.phone,
        email=student.email,
        address=student.address,
        user_id=user_id
    )
    db.add(db_student)
    db.commit()
    db.refresh(db_student)
    return db_student

# API接口
@app.post("/auth/register", response_model=UserInfo)
def register(user: UserCreate, db: Session = Depends(get_db)):
    existing_user = get_user_by_username(db, user.username)
    if existing_user:
        raise HTTPException(status_code=400, detail="用户名已存在")
    new_user = create_user(db, user)
    return new_user

@app.post("/auth/login")
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="用户名或密码错误")
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/auth/me", response_model=UserInfo)
def get_me(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    token_data = decode_access_token(token)
    if not token_data:
        raise HTTPException(status_code=401, detail="无效的访问令牌")
    user = get_user_by_username(db, token_data["username"])
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

@app.post("/students", response_model=StudentInfo)
def create_student_endpoint(student: StudentCreate, token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    token_data = decode_access_token(token)
    if not token_data:
        raise HTTPException(status_code=401, detail="无效的访问令牌")
    user = get_user_by_username(db, token_data["username"])
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    existing_student = get_student_by_no(db, student.student_no)
    if existing_student:
        raise HTTPException(status_code=400, detail="学号已存在")
    new_student = create_student(db, student, user.id)
    return new_student

@app.get("/students/{student_id}", response_model=StudentInfo)
def get_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    student = db.query(Student).filter(Student.id == student_id).first()
    if not student:
        raise HTTPException(status_code=404, detail="学生不存在")
    return student

@app.get("/students", response_model=List[StudentInfo])
def get_students_endpoint(page: int = 1, page_size: int = 10, db: Session = Depends(get_db)):
    start_index = (page - 1) * page_size
    end_index = start_index + page_size
    students = db.query(Student).offset(start_index).limit(page_size).all()
    return students

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

优缺点分析

优点

缺点

简单易上手,适合快速验证想法

代码可读性差,所有代码堆积在一个文件中

部署简单,只需要运行一个文件

耦合度高,修改一个模块时可能会影响其他模块

开发速度快,不需要考虑应用结构

测试困难,所有功能都在一个应用中,测试覆盖率难以提高


2.2 阶段二:扁平多应用(初级,简单拆分)

这是 Python 项目的初级结构,将项目的功能模块拆分为多个独立的扁平应用,每个应用负责一个特定的业务功能。

项目结构

代码语言:javascript
复制
student-management-system/
├── .github/
│   └── workflows/
│       └── ci.yml
├── apps/
│   ├── auth/
│   │   ├── __init__.py
│   │   ├── main.py  # 身份验证应用的主入口文件
│   │   ├── models.py  # 身份验证应用的数据模型
│   │   ├── schemas.py  # 身份验证应用的数据验证模型
│   │   ├── services.py  # 身份验证应用的业务逻辑
│   │   └── utils.py  # 身份验证应用的工具函数
│   └── students/
│       ├── __init__.py
│       ├── main.py  # 学生管理应用的主入口文件
│       ├── models.py  # 学生管理应用的数据模型
│       ├── schemas.py  # 学生管理应用的数据验证模型
│       ├── services.py  # 学生管理应用的业务逻辑
│       └── utils.py  # 学生管理应用的工具函数
├── core/
│   ├── __init__.py
│   ├── config.py  # 项目配置
│   ├── database.py  # 数据库连接配置
│   └── dependencies.py  # 依赖注入函数
├── tests/
│   ├── __init__.py
│   ├── test_auth.py
│   └── test_students.py
├── .dockerignore
├── .gitignore
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
└── README.md

应用间通信:在扁平多应用结构中,应用间通信可以通过以下方式:

  1. 内部调用:直接导入其他应用的业务逻辑函数;
  2. HTTP 请求:通过 HTTP 请求调用其他应用的 API 接口;
  3. 消息队列:通过消息队列(如 RabbitMQ、Redis)实现异步通信。

代码示例(内部调用)

代码语言:javascript
复制
# apps/auth/services.py
from sqlalchemy.orm import Session
from apps.auth.models import User
from apps.auth.schemas import UserCreate
from core.security import verify_password, get_password_hash

def get_user_by_username(db: Session, username: str):
    return db.query(User).filter(User.username == username).first()

def create_user(db: Session, user: UserCreate):
    hashed_password = get_password_hash(user.password)
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def authenticate_user(db: Session, username: str, password: str):
    user = get_user_by_username(db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

# apps/students/services.py
from sqlalchemy.orm import Session
from apps.students.models import Student
from apps.students.schemas import StudentCreate
from apps.auth.services import get_user_by_username

def get_student_by_no(db: Session, student_no: str):
    return db.query(Student).filter(Student.student_no == student_no).first()

def create_student(db: Session, student: StudentCreate, username: str):
    user = get_user_by_username(db, username)
    if not user:
        raise Exception("用户不存在")
    existing_student = get_student_by_no(db, student.student_no)
    if existing_student:
        raise Exception("学号已存在")
    db_student = Student(
        student_no=student.student_no,
        name=student.name,
        gender=student.gender,
        birthdate=student.birthdate,
        phone=student.phone,
        email=student.email,
        address=student.address,
        user_id=user.id
    )
    db.add(db_student)
    db.commit()
    db.refresh(db_student)
    return db_student

优缺点分析

优点

缺点

代码可读性提高,每个应用负责一个特定的业务功能

应用边界划分不清,容易出现循环导入

耦合度降低,修改一个模块时不会影响其他模块

共享组件(如配置、数据库)需要重复定义

测试困难降低,每个应用可以独立测试

部署复杂,需要部署所有应用


2.3 阶段三:分层多应用(中级,标准架构)

这是 Python 项目的标准架构,将项目的功能模块拆分为多个独立的分层应用,每个应用负责一个特定的业务功能,并且每个应用都有自己的 API、业务逻辑、数据模型、工具函数,共享组件(如配置、数据库、工具)放在项目的根目录下。

项目结构

代码语言:javascript
复制
student-management-system/
├── .github/
│   └── workflows/
│       └── ci.yml
├── apps/
│   ├── auth/
│   │   ├── __init__.py
│   │   ├── api/
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/
│   │   │   │   ├── __init__.py
│   │   │   │   ├── register.py
│   │   │   │   └── login.py
│   │   │   └── schemas/
│   │   │       ├── __init__.py
│   │   │       ├── register.py
│   │   │       └── login.py
│   │   ├── core/
│   │   │   ├── __init__.py
│   │   │   ├── config.py
│   │   │   └── security.py
│   │   ├── models/
│   │   │   ├── __init__.py
│   │   │   └── user.py
│   │   ├── services/
│   │   │   ├── __init__.py
│   │   │   └── user_service.py
│   │   └── main.py
│   └── students/
│       ├── __init__.py
│       ├── api/
│       │   ├── __init__.py
│       │   ├── endpoints/
│       │   │   ├── __init__.py
│       │   │   ├── create.py
│       │   │   ├── get.py
│       │   │   └── list.py
│       │   └── schemas/
│       │       ├── __init__.py
│       │       ├── create.py
│       │       ├── get.py
│       │       └── list.py
│       ├── core/
│       │   ├── __init__.py
│       │   └── config.py
│       ├── models/
│       │   ├── __init__.py
│       │   └── student.py
│       ├── services/
│       │   ├── __init__.py
│       │   └── student_service.py
│       └── main.py
├── shared/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py  # 项目配置
│   │   ├── database.py  # 数据库连接配置
│   │   └── dependencies.py  # 依赖注入函数
│   ├── models/
│   │   ├── __init__.py
│   │   └── base.py  # 数据模型的基类
│   ├── services/
│   │   ├── __init__.py
│   │   └── base_service.py  # 业务逻辑的基类
│   └── utils/
│       ├── __init__.py
│       ├── logger.py  # 日志管理工具
│       ├── exceptions.py  # 自定义异常类
│       └── helpers.py  # 辅助工具函数
├── tests/
│   ├── __init__.py
│   ├── test_auth/
│   │   ├── __init__.py
│   │   ├── test_register.py
│   │   └── test_login.py
│   └── test_students/
│       ├── __init__.py
│       ├── test_create.py
│       ├── test_get.py
│       └── test_list.py
├── .dockerignore
├── .gitignore
├── docker-compose.yml
├── Dockerfile.auth
├── Dockerfile.students
├── requirements.auth.txt
├── requirements.students.txt
└── README.md

应用间通信:在分层多应用结构中,应用间通信可以通过以下方式:

  1. 内部调用:直接导入其他应用的业务逻辑函数;
  2. HTTP 请求:通过 HTTP 请求调用其他应用的 API 接口;
  3. 消息队列:通过消息队列(如 RabbitMQ、Redis)实现异步通信。

代码示例(共享组件)

代码语言:javascript
复制
# shared/core/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base
from shared.core.config import settings

engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# shared/core/config.py
from pydantic import BaseSettings

class Settings(BaseSettings):
    # 项目配置
    PROJECT_NAME: str = "学生管理系统"
    PROJECT_VERSION: str = "1.0.0"
    PROJECT_DESCRIPTION: str = "一个基于FastAPI的学生管理系统"
    
    # 数据库配置
    DATABASE_URL: str = "sqlite:///./test.db"
    
    # JWT配置
    SECRET_KEY: str = "your-secret-key-here"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    
    class Config:
        env_file = ".env"

settings = Settings()

# shared/core/security.py
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from shared.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def decode_access_token(token: str):
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            return None
        return {"username": username}
    except JWTError:
        return None

部署方案:使用 Docker Compose 部署分层多应用结构的项目:

代码语言:javascript
复制
# docker-compose.yml
version: "3.8"

services:
  auth-app:
    build:
      context: .
      dockerfile: Dockerfile.auth
    ports:
      - "8001:8000"
    environment:
      - DATABASE_URL=${DATABASE_URL}
      - SECRET_KEY=${SECRET_KEY}
    depends_on:
      - db

  students-app:
    build:
      context: .
      dockerfile: Dockerfile.students
    ports:
      - "8002:8000"
    environment:
      - DATABASE_URL=${DATABASE_URL}
      - SECRET_KEY=${SECRET_KEY}
    depends_on:
      - db

  db:
    image: mysql:8.0
    container_name: mysql
    restart: always
    environment:
      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD}
      MYSQL_DATABASE: ${MYSQL_DATABASE}
    ports:
      - "3306:3306"
    volumes:
      - mysql-data:/var/lib/mysql

volumes:
  mysql-data:

优缺点分析

优点

缺点

代码可读性提高,每个应用负责一个特定的业务功能,并且每个应用都有自己的分层架构

部署复杂度提高,需要部署多个应用

耦合度降低,共享组件(如配置、数据库)放在项目的根目录下,避免重复定义

应用间通信需要通过 HTTP 请求或消息队列,增加了网络传输的开销

测试困难降低,每个应用可以独立测试,共享组件也可以独立测试

开发复杂度提高,需要考虑应用间通信的问题


2.4 阶段四:微服务雏形(高级,服务拆分)

这是 Python 项目的高级结构,将项目的功能模块拆分为多个独立的微服务,每个微服务负责一个特定的业务功能,并且每个微服务都有自己的数据库、配置、API、业务逻辑、数据模型、工具函数。

项目结构

代码语言:javascript
复制
student-management-system/
├── .github/
│   └── workflows/
│       └── ci.yml
├── services/
│   ├── auth/
│   │   ├── .github/
│   │   │   └── workflows/
│   │   │       └── ci.yml
│   │   ├── apps/
│   │   │   └── auth/
│   │   │       ├── __init__.py
│   │   │       ├── api/
│   │   │       ├── core/
│   │   │       ├── models/
│   │   │       ├── services/
│   │   │       └── main.py
│   │   ├── core/
│   │   │   ├── __init__.py
│   │   │   ├── config.py
│   │   │   └── database.py
│   │   ├── tests/
│   │   ├── .dockerignore
│   │   ├── .gitignore
│   │   ├── Dockerfile
│   │   ├── requirements.txt
│   │   └── README.md
│   └── students/
│       ├── .github/
│       │   └── workflows/
│       │       └── ci.yml
│       ├── apps/
│       │   └── students/
│       │       ├── __init__.py
│       │       ├── api/
│       │       ├── core/
│       │       ├── models/
│       │       ├── services/
│       │       └── main.py
│       ├── core/
│       │   ├── __init__.py
│       │   ├── config.py
│       │   └── database.py
│       ├── tests/
│       ├── .dockerignore
│       ├── .gitignore
│       ├── Dockerfile
│       ├── requirements.txt
│       └── README.md
├── shared/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── security.py
│   ├── models/
│   │   └── base.py
│   └── utils/
├── .gitignore
├── docker-compose.yml
└── README.md

应用间通信:在微服务雏形结构中,应用间通信主要通过 HTTP 请求或消息队列实现。

部署方案:使用 Kubernetes+Ingress 部署微服务雏形结构的项目:

代码语言:javascript
复制
# services/auth/k8s/auth-deployment.yml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: auth-deployment
  labels:
    app: auth
spec:
  replicas: 3
  selector:
    matchLabels:
      app: auth
  template:
    metadata:
      labels:
        app: auth
    spec:
      containers:
        - name: auth
          image: auth:latest
          ports:
            - containerPort: 8000
          env:
            - name: DATABASE_URL
              valueFrom:
                configMapKeyRef:
                  name: auth-config
                  key: DATABASE_URL
            - name: SECRET_KEY
              valueFrom:
                secretKeyRef:
                  name: auth-secret
                  key: SECRET_KEY
          resources:
            limits:
              memory: "512Mi"
              cpu: "500m"
            requests:
              memory: "256Mi"
              cpu: "250m"
---
apiVersion: v1
kind: Service
metadata:
  name: auth-service
spec:
  selector:
    app: auth
  ports:
    - protocol: TCP
      port: 8000
      targetPort: 8000
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: auth-config
data:
  DATABASE_URL: mysql+pymysql://root:123456@mysql/auth_db
---
apiVersion: v1
kind: Secret
metadata:
  name: auth-secret
type: Opaque
data:
  SECRET_KEY: eW91ci1zZWNyZXQta2V5LWhlcmU=

优缺点分析

优点

缺点

代码可读性提高,每个微服务负责一个特定的业务功能,并且每个微服务都有自己的数据库、配置、API、业务逻辑、数据模型、工具函数

部署复杂度高,需要部署多个微服务和相关的基础设施(如 Kubernetes、Ingress、MySQL、Redis、RabbitMQ)

耦合度降低,每个微服务都是独立的,可以独立开发、测试、部署、升级、回滚

应用间通信需要通过 HTTP 请求或消息队列,增加了网络传输的开销和延迟

测试困难降低,每个微服务可以独立测试

开发复杂度高,需要考虑应用间通信、数据一致性、服务注册与发现、负载均衡、容错、限流等问题


三、常见问题与解决方案
3.1 应用边界划分不清

问题:应用边界划分不清,导致应用间通信频繁,代码耦合度高。

解决方案:根据业务功能的职责和边界划分应用,每个应用负责一个特定的业务领域,如用户管理、学生管理、课程管理、支付管理等。

3.2 共享组件版本不兼容

问题:共享组件版本不兼容,导致不同应用之间出现代码冲突。

解决方案:使用依赖管理工具(如 poetry、pipenv)管理共享组件的版本,确保所有应用使用相同版本的共享组件。

3.3 部署复杂度高

问题:部署复杂度高,需要部署多个应用和相关的基础设施。

解决方案:使用自动化部署工具(如 Docker Compose、Kubernetes、GitLab CI/CD、GitHub Actions)简化部署流程。


四、总结与展望

通过以上步骤,我们详解了 Python 项目多应用架构的 3 种演进方案,包括单文件、扁平多应用、分层多应用、微服务雏形。多应用架构可以提高代码的可读性、可维护性、可测试性和可扩展性,但也会增加部署复杂度和开发复杂度。

在实际开发中,我们需要根据项目的规模和复杂度选择合适的应用架构。如果项目的规模较小,复杂度较低,我们可以选择单文件或扁平多应用的结构;如果项目的规模较大,复杂度较高,我们可以选择分层多应用或微服务雏形的结构。

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2026-01-12,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 摘要
    • 一、引言:为什么要做多应用结构?
    • 二、应用结构演进路径
      • 2.1 阶段一:单文件(入门级,代码堆积)
      • 2.2 阶段二:扁平多应用(初级,简单拆分)
      • 2.3 阶段三:分层多应用(中级,标准架构)
      • 2.4 阶段四:微服务雏形(高级,服务拆分)
    • 三、常见问题与解决方案
      • 3.1 应用边界划分不清
      • 3.2 共享组件版本不兼容
      • 3.3 部署复杂度高
    • 四、总结与展望
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档