当前位置：首页 > news >正文

Django+FastAPI+Vue微服务架构指南

news 2025/9/26 8:23:10

🏗️ Django+FastAPI+Vue微服务架构指南

📋 文档概述

本文档详细介绍如何构建Django+FastAPI+Vue的现代化微服务架构，实现前后端完全分离的企业级Web应用。

适用场景：

大中型企业级应用
需要复杂权限管理的系统
高并发和高性能要求的应用
团队协作开发项目

技术栈：

用户服务：Django + DRF + PostgreSQL
业务服务：FastAPI + SQLAlchemy + Redis
前端界面：Vue 3 + TypeScript + Vite
基础设施：Docker + Nginx + API Gateway

🏗️ 微服务架构设计

📊 整体架构图

                    ┌─────────────────┐│   Vue 3 前端     ││                 ││ • 用户界面       ││ • 状态管理       ││ • 路由导航       │└─────────┬───────┘│ HTTP/WebSocket┌─────────▼───────┐│  API Gateway    ││  (Nginx/Kong)   │└─────────┬───────┘│┌─────────────────┼─────────────────┐│                 │                 │┌───────▼──────┐  ┌───────▼──────┐  ┌──────▼──────┐│ Django服务    │  │ FastAPI服务   │  │  其他服务    ││              │  │              │  │             ││ • 用户管理    │  │ • 高性能API   │  │ • 通知服务   ││ • 权限系统    │  │ • 实时功能    │  │ • 支付服务   ││ • Admin后台   │  │ • 文件处理    │  │ • ...       ││ • 复杂业务    │  │ • 数据分析    │  │             │└───────┬──────┘  └───────┬──────┘  └──────┬──────┘│                 │                 │┌───────▼──────┐  ┌───────▼──────┐  ┌──────▼──────┐│ PostgreSQL   │  │ Redis + DB   │  │   MongoDB   ││   主数据库    │  │  缓存+存储   │  │  文档数据库  │└──────────────┘  └──────────────┘  └─────────────┘

🎯 微服务划分原则

按业务功能划分：

Django用户服务：认证、授权、用户管理、权限系统
FastAPI业务服务：高性能API、实时功能、数据处理
Vue前端服务：用户界面、交互逻辑、状态管理

服务特点：

✅ 独立开发：每个服务可以独立开发和部署
✅ 技术异构：使用最适合的技术栈
✅ 数据隔离：每个服务管理自己的数据
✅ 松耦合：通过API进行服务间通信

📁 项目结构

🗂️ 微服务项目结构

microservice-webapp/
├── frontend/                    # Vue前端服务
│   ├── src/
│   │   ├── services/
│   │   │   ├── djangoApi.ts    # Django服务API
│   │   │   ├── fastApi.ts      # FastAPI服务API
│   │   │   └── apiGateway.ts   # 统一API网关
│   │   ├── stores/
│   │   │   ├── auth.ts         # 认证状态
│   │   │   ├── user.ts         # 用户数据
│   │   │   └── business.ts     # 业务数据
│   │   └── views/
│   ├── package.json
│   └── Dockerfile
├── django-service/              # Django用户服务
│   ├── config/
│   │   ├── settings.py
│   │   ├── urls.py
│   │   └── wsgi.py
│   ├── apps/
│   │   ├── authentication/     # 认证模块
│   │   ├── users/             # 用户管理
│   │   ├── permissions/       # 权限系统
│   │   └── admin_panel/       # 管理后台
│   ├── requirements.txt
│   └── Dockerfile
├── fastapi-service/             # FastAPI业务服务
│   ├── app/
│   │   ├── api/               # API端点
│   │   ├── services/          # 业务逻辑
│   │   ├── models/            # 数据模型
│   │   └── core/              # 核心配置
│   ├── requirements.txt
│   └── Dockerfile
├── api-gateway/                 # API网关配置
│   ├── nginx.conf
│   └── kong.yml
├── docker-compose.yml           # 容器编排
├── docker-compose.prod.yml      # 生产环境
└── README.md

🔧 Django用户服务开发

📝 Django项目配置

1. 项目初始化

# django-service/config/settings.py
import os
from pathlib import PathBASE_DIR = Path(__file__).resolve().parent.parent# Django REST Framework配置
INSTALLED_APPS = ['django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles',# 第三方应用'rest_framework','rest_framework_simplejwt','corsheaders','django_extensions',# 本地应用'apps.authentication','apps.users','apps.permissions',
]MIDDLEWARE = ['corsheaders.middleware.CorsMiddleware','django.middleware.security.SecurityMiddleware','django.contrib.sessions.middleware.SessionMiddleware','django.middleware.common.CommonMiddleware','django.middleware.csrf.CsrfViewMiddleware','django.contrib.auth.middleware.AuthenticationMiddleware','django.contrib.messages.middleware.MessageMiddleware','django.middleware.clickjacking.XFrameOptionsMiddleware',
]# REST Framework配置
REST_FRAMEWORK = {'DEFAULT_AUTHENTICATION_CLASSES': ['rest_framework_simplejwt.authentication.JWTAuthentication',],'DEFAULT_PERMISSION_CLASSES': ['rest_framework.permissions.IsAuthenticated',],'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination','PAGE_SIZE': 20,
}# JWT配置
from datetime import timedelta
SIMPLE_JWT = {'ACCESS_TOKEN_LIFETIME': timedelta(hours=1),'REFRESH_TOKEN_LIFETIME': timedelta(days=7),'ROTATE_REFRESH_TOKENS': True,
}# CORS配置
CORS_ALLOWED_ORIGINS = ["http://localhost:3000",  # Vue开发服务器"http://localhost:8080",  # Vue生产服务器
]# 数据库配置
DATABASES = {'default': {'ENGINE': 'django.db.backends.postgresql','NAME': os.getenv('DB_NAME', 'django_db'),'USER': os.getenv('DB_USER', 'django_user'),'PASSWORD': os.getenv('DB_PASSWORD', 'django_pass'),'HOST': os.getenv('DB_HOST', 'localhost'),'PORT': os.getenv('DB_PORT', '5432'),}
}

2. 用户模型扩展

# django-service/apps/users/models.py
from django.contrib.auth.models import AbstractUser
from django.db import models
from django.utils import timezoneclass User(AbstractUser):"""扩展的用户模型"""email = models.EmailField(unique=True)phone = models.CharField(max_length=20, blank=True)avatar = models.URLField(blank=True)department = models.CharField(max_length=100, blank=True)position = models.CharField(max_length=100, blank=True)is_verified = models.BooleanField(default=False)last_login_ip = models.GenericIPAddressField(null=True, blank=True)created_at = models.DateTimeField(default=timezone.now)updated_at = models.DateTimeField(auto_now=True)USERNAME_FIELD = 'email'REQUIRED_FIELDS = ['username', 'first_name', 'last_name']class Meta:db_table = 'users'verbose_name = '用户'verbose_name_plural = '用户管理'class UserProfile(models.Model):"""用户详细资料"""user = models.OneToOneField(User, on_delete=models.CASCADE, related_name='profile')bio = models.TextField(blank=True)birth_date = models.DateField(null=True, blank=True)location = models.CharField(max_length=100, blank=True)website = models.URLField(blank=True)social_links = models.JSONField(default=dict, blank=True)preferences = models.JSONField(default=dict, blank=True)class Meta:db_table = 'user_profiles'verbose_name = '用户资料'verbose_name_plural = '用户资料管理'

3. 权限系统

# django-service/apps/permissions/models.py
from django.db import models
from django.contrib.auth.models import Group
from apps.users.models import Userclass Permission(models.Model):"""自定义权限模型"""name = models.CharField(max_length=100, unique=True)codename = models.CharField(max_length=100, unique=True)description = models.TextField(blank=True)module = models.CharField(max_length=50)  # 权限所属模块created_at = models.DateTimeField(auto_now_add=True)class Meta:db_table = 'custom_permissions'verbose_name = '权限'verbose_name_plural = '权限管理'class Role(models.Model):"""角色模型"""name = models.CharField(max_length=100, unique=True)description = models.TextField(blank=True)permissions = models.ManyToManyField(Permission, blank=True)is_active = models.BooleanField(default=True)created_at = models.DateTimeField(auto_now_add=True)class Meta:db_table = 'roles'verbose_name = '角色'verbose_name_plural = '角色管理'class UserRole(models.Model):"""用户角色关联"""user = models.ForeignKey(User, on_delete=models.CASCADE)role = models.ForeignKey(Role, on_delete=models.CASCADE)assigned_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name='assigned_roles')assigned_at = models.DateTimeField(auto_now_add=True)expires_at = models.DateTimeField(null=True, blank=True)class Meta:db_table = 'user_roles'unique_together = ['user', 'role']verbose_name = '用户角色'verbose_name_plural = '用户角色管理'

4. API视图

# django-service/apps/authentication/views.py
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework_simplejwt.views import TokenObtainPairView
from django.contrib.auth import authenticate
import requestsclass CustomTokenObtainPairView(TokenObtainPairView):"""自定义JWT登录"""def post(self, request, *args, **kwargs):response = super().post(request, *args, **kwargs)if response.status_code == 200:# 登录成功后，通知FastAPI服务更新用户状态user_email = request.data.get('email')try:# 调用FastAPI服务的用户状态更新接口fastapi_response = requests.post('http://fastapi-service:8000/api/auth/login-notify',json={'email': user_email, 'action': 'login'},timeout=5)except requests.exceptions.RequestException:# FastAPI服务不可用时，不影响Django的登录流程passreturn response@api_view(['POST'])
@permission_classes([AllowAny])
def register(request):"""用户注册"""serializer = UserCreateSerializer(data=request.data)if serializer.is_valid():user = serializer.save()# 通知FastAPI服务创建用户相关数据try:requests.post('http://fastapi-service:8000/api/users/create-notify',json={'user_id': user.id, 'email': user.email},timeout=5)except requests.exceptions.RequestException:passreturn Response({'message': '注册成功','user_id': user.id}, status=status.HTTP_201_CREATED)return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

5. Admin管理后台

# django-service/apps/users/admin.py
from django.contrib import admin
from django.contrib.auth.admin import UserAdmin as BaseUserAdmin
from .models import User, UserProfile@admin.register(User)
class UserAdmin(BaseUserAdmin):"""用户管理后台"""list_display = ['email', 'username', 'first_name', 'last_name', 'is_active', 'is_verified', 'created_at']list_filter = ['is_active', 'is_verified', 'is_staff', 'created_at']search_fields = ['email', 'username', 'first_name', 'last_name']ordering = ['-created_at']fieldsets = BaseUserAdmin.fieldsets + (('额外信息', {'fields': ('phone', 'avatar', 'department', 'position', 'is_verified', 'last_login_ip')}),('时间信息', {'fields': ('created_at', 'updated_at')}),)readonly_fields = ['created_at', 'updated_at', 'last_login_ip']@admin.register(UserProfile)
class UserProfileAdmin(admin.ModelAdmin):"""用户资料管理后台"""list_display = ['user', 'location', 'birth_date']search_fields = ['user__email', 'user__username', 'location']list_filter = ['birth_date', 'location']

⚡ FastAPI业务服务开发

📝 FastAPI服务配置

1. 应用结构

# fastapi-service/app/main.py
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from app.api.endpoints import analytics, files, notifications
from app.core.config import settings
from app.db.database import engine, Base
import httpxapp = FastAPI(title="业务处理服务",description="高性能业务逻辑处理和数据分析服务",version="1.0.0"
)# CORS配置
app.add_middleware(CORSMiddleware,allow_origins=settings.ALLOWED_HOSTS,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)# 创建数据库表
@app.on_event("startup")
async def startup_event():Base.metadata.create_all(bind=engine)# 包含路由
app.include_router(analytics.router, prefix="/api/analytics", tags=["数据分析"])
app.include_router(files.router, prefix="/api/files", tags=["文件处理"])
app.include_router(notifications.router, prefix="/api/notifications", tags=["通知服务"])# 与Django服务通信的工具函数
async def verify_user_with_django(token: str):"""向Django服务验证用户token"""async with httpx.AsyncClient() as client:try:response = await client.post(f"{settings.DJANGO_SERVICE_URL}/api/auth/verify-token/",headers={"Authorization": f"Bearer {token}"},timeout=5.0)if response.status_code == 200:return response.json()except httpx.RequestError:passreturn None

2. 数据分析服务

# fastapi-service/app/api/endpoints/analytics.py
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.services.analytics import AnalyticsService
from app.models.analytics import UserActivity, BusinessMetrics
import asynciorouter = APIRouter()@router.get("/user-stats/{user_id}")
async def get_user_statistics(user_id: int,db: Session = Depends(get_db)
):"""获取用户统计数据"""analytics_service = AnalyticsService(db)# 高性能数据聚合stats = await analytics_service.calculate_user_stats(user_id)return {"user_id": user_id,"total_activities": stats["activities"],"login_frequency": stats["login_freq"],"feature_usage": stats["features"],"performance_metrics": stats["performance"]}@router.post("/track-activity")
async def track_user_activity(activity_data: dict,background_tasks: BackgroundTasks,db: Session = Depends(get_db)
):"""异步追踪用户活动"""# 后台任务处理，不阻塞响应background_tasks.add_task(process_activity_data,activity_data,db)return {"status": "活动数据已提交处理"}async def process_activity_data(activity_data: dict, db: Session):"""后台处理用户活动数据"""# 复杂的数据处理逻辑activity = UserActivity(user_id=activity_data["user_id"],action=activity_data["action"],metadata=activity_data.get("metadata", {}),ip_address=activity_data.get("ip"),user_agent=activity_data.get("user_agent"))db.add(activity)db.commit()# 实时数据分析await update_real_time_metrics(activity_data["user_id"], db)@router.get("/real-time-dashboard")
async def get_real_time_dashboard():"""实时数据看板"""# 使用Redis缓存的实时数据from app.core.redis import redis_clientdashboard_data = await redis_client.get("dashboard:real_time")if dashboard_data:return json.loads(dashboard_data)# 如果缓存不存在，计算并缓存data = await calculate_dashboard_metrics()await redis_client.setex("dashboard:real_time", 60, json.dumps(data))return data

3. 文件处理服务

# fastapi-service/app/api/endpoints/files.py
from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse
import aiofiles
import asyncio
from app.services.file_processor import FileProcessor
from app.core.config import settingsrouter = APIRouter()@router.post("/upload-batch")
async def upload_multiple_files(files: list[UploadFile] = File(...),background_tasks: BackgroundTasks = BackgroundTasks()
):"""批量文件上传和处理"""if len(files) > 10:raise HTTPException(status_code=400, detail="最多支持10个文件同时上传")file_processor = FileProcessor()upload_results = []for file in files:# 验证文件类型和大小if file.size > settings.MAX_FILE_SIZE:upload_results.append({"filename": file.filename,"status": "error","message": "文件大小超过限制"})continue# 异步保存文件file_path = await file_processor.save_file(file)# 后台处理文件（图片压缩、文档转换等）background_tasks.add_task(file_processor.process_file_async,file_path,file.content_type)upload_results.append({"filename": file.filename,"status": "success","file_path": file_path,"size": file.size})return {"results": upload_results}@router.get("/download/{file_id}")
async def download_file(file_id: str):"""流式文件下载"""file_processor = FileProcessor()file_info = await file_processor.get_file_info(file_id)if not file_info:raise HTTPException(status_code=404, detail="文件不存在")# 流式响应，支持大文件下载async def file_streamer():async with aiofiles.open(file_info["path"], "rb") as file:while chunk := await file.read(8192):  # 8KB chunksyield chunkreturn StreamingResponse(file_streamer(),media_type=file_info["content_type"],headers={"Content-Disposition": f"attachment; filename={file_info['filename']}"})

🎨 Vue前端统一界面

📝 多服务API管理

1. API网关服务

// frontend/src/services/apiGateway.ts
import axios, { AxiosInstance } from 'axios'
import { useAuthStore } from '@/stores/auth'class ApiGateway {private djangoApi: AxiosInstanceprivate fastApi: AxiosInstanceconstructor() {// Django用户服务APIthis.djangoApi = axios.create({baseURL: import.meta.env.VITE_DJANGO_API_URL || 'http://localhost:8001',timeout: 10000,})// FastAPI业务服务APIthis.fastApi = axios.create({baseURL: import.meta.env.VITE_FASTAPI_URL || 'http://localhost:8002',timeout: 10000,})this.setupInterceptors()}private setupInterceptors() {// 统一的请求拦截器const requestInterceptor = (config: any) => {const authStore = useAuthStore()if (authStore.token) {config.headers.Authorization = `Bearer ${authStore.token}`}return config}// 统一的响应拦截器const responseErrorInterceptor = (error: any) => {if (error.response?.status === 401) {const authStore = useAuthStore()authStore.logout()window.location.href = '/login'}return Promise.reject(error)}// 应用拦截器this.djangoApi.interceptors.request.use(requestInterceptor)this.fastApi.interceptors.request.use(requestInterceptor)this.djangoApi.interceptors.response.use(response => response,responseErrorInterceptor)this.fastApi.interceptors.response.use(response => response,responseErrorInterceptor)}// 用户相关API（Django服务）get userService() {return {login: (credentials: any) => this.djangoApi.post('/api/auth/login/', credentials),register: (userData: any) => this.djangoApi.post('/api/auth/register/', userData),getProfile: () => this.djangoApi.get('/api/auth/me/'),updateProfile: (data: any) => this.djangoApi.patch('/api/auth/me/', data),getUsers: (params: any) => this.djangoApi.get('/api/users/', { params }),createUser: (userData: any) => this.djangoApi.post('/api/users/', userData),updateUser: (id: number, data: any) => this.djangoApi.patch(`/api/users/${id}/`, data),deleteUser: (id: number) => this.djangoApi.delete(`/api/users/${id}/`),}}// 业务相关API（FastAPI服务）get businessService() {return {getUserStats: (userId: number) => this.fastApi.get(`/api/analytics/user-stats/${userId}`),trackActivity: (activityData: any) => this.fastApi.post('/api/analytics/track-activity', activityData),getRealTimeDashboard: () => this.fastApi.get('/api/analytics/real-time-dashboard'),uploadFiles: (files: FormData) => this.fastApi.post('/api/files/upload-batch', files),downloadFile: (fileId: string) => this.fastApi.get(`/api/files/download/${fileId}`, { responseType: 'blob' }),sendNotification: (notificationData: any) => this.fastApi.post('/api/notifications/send', notificationData),}}
}export const apiGateway = new ApiGateway()

2. 统一状态管理

// frontend/src/stores/app.ts
import { defineStore } from 'pinia'
import { ref, computed } from 'vue'
import { apiGateway } from '@/services/apiGateway'
import type { User, DashboardData } from '@/types'export const useAppStore = defineStore('app', () => {// 状态const user = ref<User | null>(null)const dashboardData = ref<DashboardData | null>(null)const isLoading = ref(false)const notifications = ref<any[]>([])// 计算属性const userStats = computed(() => {if (!dashboardData.value) return nullreturn dashboardData.value.userStats})// 用户相关操作async function fetchUserProfile() {try {isLoading.value = trueconst response = await apiGateway.userService.getProfile()user.value = response.data// 同时获取用户统计数据（FastAPI服务）if (user.value?.id) {await fetchUserStats(user.value.id)}} catch (error) {console.error('获取用户信息失败:', error)} finally {isLoading.value = false}}// 业务数据操作async function fetchUserStats(userId: number) {try {const response = await apiGateway.businessService.getUserStats(userId)if (dashboardData.value) {dashboardData.value.userStats = response.data} else {dashboardData.value = { userStats: response.data }}} catch (error) {console.error('获取用户统计失败:', error)}}// 实时数据更新async function fetchRealTimeDashboard() {try {const response = await apiGateway.businessService.getRealTimeDashboard()dashboardData.value = { ...dashboardData.value, ...response.data }} catch (error) {console.error('获取实时数据失败:', error)}}// 文件操作async function uploadFiles(files: FileList) {try {isLoading.value = trueconst formData = new FormData()Array.from(files).forEach(file => {formData.append('files', file)})const response = await apiGateway.businessService.uploadFiles(formData)// 追踪用户活动await apiGateway.businessService.trackActivity({user_id: user.value?.id,action: 'file_upload',metadata: { file_count: files.length }})return response.data} catch (error) {console.error('文件上传失败:', error)throw error} finally {isLoading.value = false}}return {user,dashboardData,isLoading,notifications,userStats,fetchUserProfile,fetchUserStats,fetchRealTimeDashboard,uploadFiles}
})

3. 智能组件示例

<!-- frontend/src/components/DashboardWidget.vue -->
<template><div class="dashboard-widget"><el-card class="widget-card"><template #header><div class="card-header"><span>实时数据看板</span><el-button type="text" :loading="refreshing" @click="refreshData"><el-icon><Refresh /></el-icon></el-button></div></template><!-- 用户统计 (来自FastAPI) --><div class="stats-grid"><div class="stat-item"><div class="stat-value">{{ userStats?.total_activities || 0 }}</div><div class="stat-label">总活动数</div></div><div class="stat-item"><div class="stat-value">{{ userStats?.login_frequency || 0 }}</div><div class="stat-label">登录频率</div></div></div><!-- 实时图表 --><div class="chart-container"><RealTimeChart :data="dashboardData?.chartData" /></div><!-- 用户操作 (调用Django) --><div class="actions"><el-button @click="showUserManagement">用户管理</el-button><el-button @click="exportData">导出数据</el-button></div></el-card></div>
</template><script setup lang="ts">
import { ref, onMounted, onUnmounted } from 'vue'
import { useAppStore } from '@/stores/app'
import { apiGateway } from '@/services/apiGateway'
import RealTimeChart from './RealTimeChart.vue'const appStore = useAppStore()
const refreshing = ref(false)
let refreshTimer: number | null = null// 计算属性
const { userStats, dashboardData } = storeToRefs(appStore)// 刷新数据
async function refreshData() {refreshing.value = truetry {await Promise.all([appStore.fetchRealTimeDashboard(),appStore.fetchUserProfile()])} finally {refreshing.value = false}
}// 显示用户管理（Django Admin）
function showUserManagement() {// 打开Django Admin后台window.open('/admin/', '_blank')
}// 导出数据（FastAPI处理）
async function exportData() {try {const response = await apiGateway.businessService.exportUserData()// 处理文件下载const blob = new Blob([response.data])const url = window.URL.createObjectURL(blob)const a = document.createElement('a')a.href = urla.download = 'user_data_export.xlsx'a.click()window.URL.revokeObjectURL(url)} catch (error) {ElMessage.error('导出失败')}
}// 自动刷新
onMounted(() => {refreshData()// 每30秒自动刷新实时数据refreshTimer = setInterval(() => {appStore.fetchRealTimeDashboard()}, 30000)
})onUnmounted(() => {if (refreshTimer) {clearInterval(refreshTimer)}
})
</script>

🐳 容器化部署

📦 Docker Compose配置

# docker-compose.yml
version: '3.8'services:# PostgreSQL - Django数据库django-db:image: postgres:15environment:POSTGRES_DB: django_dbPOSTGRES_USER: django_userPOSTGRES_PASSWORD: django_passvolumes:- django_postgres_data:/var/lib/postgresql/dataports:- "5432:5432"# Redis - FastAPI缓存redis:image: redis:7-alpineports:- "6379:6379"volumes:- redis_data:/data# Django用户服务django-service:build: ./django-serviceports:- "8001:8000"environment:- DEBUG=True- DB_HOST=django-db- DB_NAME=django_db- DB_USER=django_user- DB_PASSWORD=django_pass- FASTAPI_SERVICE_URL=http://fastapi-service:8000depends_on:- django-dbvolumes:- ./django-service:/appcommand: python manage.py runserver 0.0.0.0:8000# FastAPI业务服务fastapi-service:build: ./fastapi-serviceports:- "8002:8000"environment:- DJANGO_SERVICE_URL=http://django-service:8000- REDIS_URL=redis://redis:6379- DATABASE_URL=postgresql://fastapi_user:fastapi_pass@fastapi-db:5432/fastapi_dbdepends_on:- redis- fastapi-dbvolumes:- ./fastapi-service:/appcommand: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload# FastAPI专用数据库fastapi-db:image: postgres:15environment:POSTGRES_DB: fastapi_dbPOSTGRES_USER: fastapi_userPOSTGRES_PASSWORD: fastapi_passvolumes:- fastapi_postgres_data:/var/lib/postgresql/dataports:- "5433:5432"# Vue前端frontend:build: ./frontendports:- "3000:3000"environment:- VITE_DJANGO_API_URL=http://localhost:8001- VITE_FASTAPI_URL=http://localhost:8002volumes:- ./frontend:/app- /app/node_modulescommand: npm run dev -- --host 0.0.0.0# Nginx API网关nginx:image: nginx:alpineports:- "80:80"volumes:- ./nginx.conf:/etc/nginx/nginx.confdepends_on:- django-service- fastapi-service- frontendvolumes:django_postgres_data:fastapi_postgres_data:redis_data:

🌐 Nginx API网关配置

# nginx.conf
events {worker_connections 1024;
}http {upstream django_backend {server django-service:8000;}upstream fastapi_backend {server fastapi-service:8000;}upstream frontend_app {server frontend:3000;}# 负载均衡和健康检查server {listen 80;# Django用户服务路由location /api/auth/ {proxy_pass http://django_backend;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;}location /api/users/ {proxy_pass http://django_backend;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;}location /admin/ {proxy_pass http://django_backend;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;}# FastAPI业务服务路由location /api/analytics/ {proxy_pass http://fastapi_backend;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;}location /api/files/ {proxy_pass http://fastapi_backend;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;# 文件上传大小限制client_max_body_size 100M;}location /api/notifications/ {proxy_pass http://fastapi_backend;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;}# FastAPI文档location /docs {proxy_pass http://fastapi_backend;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;}# Vue前端应用location / {proxy_pass http://frontend_app;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;# 支持Vue Router的History模式try_files $uri $uri/ /index.html;}# WebSocket支持（如果需要）location /ws/ {proxy_pass http://fastapi_backend;proxy_http_version 1.1;proxy_set_header Upgrade $http_upgrade;proxy_set_header Connection "upgrade";proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;}}
}

🚀 服务间通信

📡 同步通信

# Django调用FastAPI
import httpx
from django.conf import settingsclass FastAPIClient:def __init__(self):self.base_url = settings.FASTAPI_SERVICE_URLself.timeout = 10.0async def notify_user_login(self, user_id: int, email: str):"""通知FastAPI用户登录"""async with httpx.AsyncClient() as client:try:response = await client.post(f"{self.base_url}/api/auth/login-notify",json={"user_id": user_id, "email": email},timeout=self.timeout)return response.json() if response.status_code == 200 else Noneexcept httpx.RequestError:return Noneasync def get_user_analytics(self, user_id: int):"""获取用户分析数据"""async with httpx.AsyncClient() as client:try:response = await client.get(f"{self.base_url}/api/analytics/user-stats/{user_id}",timeout=self.timeout)return response.json() if response.status_code == 200 else Noneexcept httpx.RequestError:return None# FastAPI调用Django
class DjangoClient:def __init__(self):self.base_url = settings.DJANGO_SERVICE_URLself.timeout = 10.0async def verify_user_token(self, token: str):"""向Django验证用户token"""async with httpx.AsyncClient() as client:try:response = await client.post(f"{self.base_url}/api/auth/verify-token/",headers={"Authorization": f"Bearer {token}"},timeout=self.timeout)return response.json() if response.status_code == 200 else Noneexcept httpx.RequestError:return None

📨 异步通信（消息队列）

# 使用Redis作为消息队列
import redis
import json
from typing import Dict, Anyclass MessageQueue:def __init__(self):self.redis_client = redis.Redis(host='redis', port=6379, db=0)def publish_event(self, event_type: str, data: Dict[Any, Any]):"""发布事件"""message = {"event_type": event_type,"data": data,"timestamp": datetime.utcnow().isoformat()}self.redis_client.publish("microservices_events", json.dumps(message))def subscribe_events(self):"""订阅事件"""pubsub = self.redis_client.pubsub()pubsub.subscribe("microservices_events")for message in pubsub.listen():if message["type"] == "message":event_data = json.loads(message["data"])self.handle_event(event_data)def handle_event(self, event_data: Dict[Any, Any]):"""处理事件"""event_type = event_data["event_type"]data = event_data["data"]if event_type == "user_registered":# 处理用户注册事件self.handle_user_registered(data)elif event_type == "file_uploaded":# 处理文件上传事件self.handle_file_uploaded(data)# Django服务发布事件
class UserService:def __init__(self):self.mq = MessageQueue()def create_user(self, user_data):user = User.objects.create(**user_data)# 发布用户创建事件self.mq.publish_event("user_registered", {"user_id": user.id,"email": user.email,"name": user.get_full_name()})return user# FastAPI服务监听事件
class EventHandler:def __init__(self):self.mq = MessageQueue()def handle_user_registered(self, data):"""处理用户注册事件"""# 为新用户创建分析记录user_analytics = UserAnalytics(user_id=data["user_id"],email=data["email"],created_at=datetime.utcnow())# 保存到FastAPI的数据库db.add(user_analytics)db.commit()

📈 监控和日志

📊 应用监控

# 统一日志配置
import structlog
import logging.configLOGGING_CONFIG = {"version": 1,"disable_existing_loggers": False,"formatters": {"json": {"()": structlog.stdlib.ProcessorFormatter,"processor": structlog.dev.ConsoleRenderer(colors=False),},},"handlers": {"default": {"level": "INFO","class": "logging.StreamHandler","formatter": "json",},"file": {"level": "INFO","class": "logging.handlers.RotatingFileHandler","filename": "app.log","maxBytes": 10485760,  # 10MB"backupCount": 5,"formatter": "json",},},"loggers": {"": {"handlers": ["default", "file"],"level": "INFO","propagate": True,},},
}logging.config.dictConfig(LOGGING_CONFIG)
structlog.configure(processors=[structlog.stdlib.filter_by_level,structlog.stdlib.add_logger_name,structlog.stdlib.add_log_level,structlog.stdlib.PositionalArgumentsFormatter(),structlog.processors.TimeStamper(fmt="iso"),structlog.processors.StackInfoRenderer(),structlog.processors.format_exc_info,structlog.processors.UnicodeDecoder(),structlog.processors.JSONRenderer()],context_class=dict,logger_factory=structlog.stdlib.LoggerFactory(),wrapper_class=structlog.stdlib.BoundLogger,cache_logger_on_first_use=True,
)logger = structlog.get_logger()# 在各服务中使用
@app.middleware("http")
async def logging_middleware(request: Request, call_next):start_time = time.time()logger.info("request_started",method=request.method,url=str(request.url),user_agent=request.headers.get("user-agent"))response = await call_next(request)process_time = time.time() - start_timelogger.info("request_completed",method=request.method,url=str(request.url),status_code=response.status_code,process_time=process_time)return response

🔍 性能监控

# 使用Prometheus监控
from prometheus_client import Counter, Histogram, generate_latest# 定义监控指标
REQUEST_COUNT = Counter('app_requests_total', 'Total app requests', ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('app_request_duration_seconds', 'Request latency')@app.middleware("http")
async def prometheus_middleware(request: Request, call_next):start_time = time.time()response = await call_next(request)# 记录指标REQUEST_COUNT.labels(method=request.method,endpoint=request.url.path,status=response.status_code).inc()REQUEST_LATENCY.observe(time.time() - start_time)return response@app.get("/metrics")
async def metrics():"""Prometheus监控指标"""return Response(generate_latest(), media_type="text/plain")