java-template/docs/redis_stream_migration_guide.md

Redis Stream 架构迁移实施指南

🎯 迁移总览

迁移目标

• 将现有内存队列批处理机制迁移到 Redis Stream
• 提升系统吞吐量从 1,400 QPS 到 5,000+ QPS
• 支持设备并发数从 2,000 提升到 10,000+
• 降低响应延迟从 2s 到 150ms

迁移策略: 渐进式零风险切换

Phase 1: 基础设施准备 + 并行验证 (Week 1-2)
Phase 2: 灰度测试 (Week 3)  
Phase 3: 全量切换 (Week 4)

---

📋 Phase 1: 基础设施准备 (Week 1-2)

Step 1.1: Redis Stream 基础设施部署

1.1.1 创建 Stream 管理器

# 创建新文件
touch /Users/brunogao/work/codes/93/release/ljwx-bigscreen/bigscreen/bigScreen/redis_stream_manager.py

📄 redis_stream_manager.py 完整代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Redis Stream 管理器
提供Stream的生产者和消费者统一接口
"""

import redis
import json
import time
import uuid
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from datetime import datetime
import asyncio
from concurrent.futures import ThreadPoolExecutor

logger = logging.getLogger(__name__)

@dataclass
class StreamMessage:
    """Stream消息数据类"""
    stream_id: str
    timestamp: int
    payload: Dict[str, Any]
    metadata: Dict[str, Any] = None

class RedisStreamManager:
    """Redis Stream统一管理器"""
    
    def __init__(self, 
                 redis_host='localhost', 
                 redis_port=6379, 
                 redis_password=None,
                 redis_db=0):
        
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port, 
            password=redis_password,
            db=redis_db,
            decode_responses=True,
            socket_keepalive=True,
            socket_keepalive_options={},
            health_check_interval=30
        )
        
        # Stream配置
        self.streams_config = {
            'health_data_stream': {
                'consumer_group': 'health_processors',
                'max_len': 100000,  # 保留最近10万条消息
                'ttl': 86400 * 7   # 7天TTL
            },
            'device_info_stream': {
                'consumer_group': 'device_processors',
                'max_len': 50000,
                'ttl': 86400 * 30  # 30天TTL
            },
            'common_event_stream': {
                'consumer_group': 'event_processors', 
                'max_len': 200000,
                'ttl': 86400 * 3   # 3天TTL
            }
        }
        
        # 初始化Stream和消费者组
        self._initialize_streams()
        
        # 性能统计
        self.stats = {
            'messages_produced': 0,
            'messages_consumed': 0,
            'errors': 0,
            'last_error_time': None
        }
        
        logger.info("🚀 RedisStreamManager初始化完成")
    
    def _initialize_streams(self):
        """初始化Stream和消费者组"""
        for stream_name, config in self.streams_config.items():
            try:
                # 创建消费者组(如果不存在)
                self.redis_client.xgroup_create(
                    stream_name, 
                    config['consumer_group'], 
                    '$', 
                    mkstream=True
                )
                logger.info(f"✅ Stream初始化: {stream_name}")
            except redis.ResponseError as e:
                if "BUSYGROUP" in str(e):
                    logger.info(f"📝 消费者组已存在: {stream_name}:{config['consumer_group']}")
                else:
                    logger.error(f"❌ Stream初始化失败: {stream_name}, error: {e}")
    
    def add_to_stream(self, 
                     stream_name: str, 
                     data: Dict[str, Any], 
                     max_len: Optional[int] = None) -> str:
        """
        添加消息到Stream
        
        Args:
            stream_name: Stream名称
            data: 消息数据
            max_len: Stream最大长度(可选)
            
        Returns:
            消息ID
        """
        try:
            # 准备消息字段
            message_fields = {
                'timestamp': int(time.time() * 1000),  # 毫秒时间戳
                'uuid': str(uuid.uuid4()),
                'payload': json.dumps(data, ensure_ascii=False)
            }
            
            # 添加元数据
            if 'device_sn' in data:
                message_fields['device_sn'] = data['device_sn']
            if 'message_type' in data:
                message_fields['message_type'] = data['message_type']
                
            # 使用配置的max_len或传入的max_len
            if max_len is None:
                max_len = self.streams_config.get(stream_name, {}).get('max_len', 10000)
            
            # 添加消息到Stream
            message_id = self.redis_client.xadd(
                stream_name,
                message_fields,
                maxlen=max_len,
                approximate=True  # 使用近似长度，性能更好
            )
            
            self.stats['messages_produced'] += 1
            logger.debug(f"✅ 消息添加成功: {stream_name}:{message_id}")
            
            return message_id
            
        except Exception as e:
            self.stats['errors'] += 1
            self.stats['last_error_time'] = datetime.now()
            logger.error(f"❌ 添加消息失败: {stream_name}, error: {e}")
            raise
    
    def add_health_data(self, health_data: Dict[str, Any]) -> str:
        """添加健康数据到健康数据Stream"""
        return self.add_to_stream('health_data_stream', health_data)
    
    def add_device_info(self, device_info: Dict[str, Any]) -> str:
        """添加设备信息到设备信息Stream"""
        return self.add_to_stream('device_info_stream', device_info)
    
    def add_common_event(self, event_data: Dict[str, Any]) -> str:
        """添加通用事件到事件Stream"""
        return self.add_to_stream('common_event_stream', event_data)
    
    def read_messages(self, 
                     stream_name: str, 
                     consumer_group: str,
                     consumer_name: str,
                     count: int = 100,
                     block: int = 1000) -> List[StreamMessage]:
        """
        从Stream读取消息
        
        Args:
            stream_name: Stream名称
            consumer_group: 消费者组
            consumer_name: 消费者名称  
            count: 读取消息数量
            block: 阻塞时间(ms)
            
        Returns:
            消息列表
        """
        try:
            response = self.redis_client.xreadgroup(
                consumer_group,
                consumer_name,
                {stream_name: '>'},
                count=count,
                block=block
            )
            
            messages = []
            if response:
                for stream, msgs in response:
                    for msg_id, fields in msgs:
                        try:
                            # 解析消息
                            payload = json.loads(fields.get('payload', '{}'))
                            metadata = {
                                'device_sn': fields.get('device_sn'),
                                'message_type': fields.get('message_type'),
                                'uuid': fields.get('uuid')
                            }
                            
                            message = StreamMessage(
                                stream_id=msg_id,
                                timestamp=int(fields.get('timestamp', 0)),
                                payload=payload,
                                metadata=metadata
                            )
                            messages.append(message)
                            
                        except Exception as e:
                            logger.error(f"❌ 解析消息失败: {msg_id}, error: {e}")
            
            if messages:
                self.stats['messages_consumed'] += len(messages)
                logger.debug(f"📥 读取消息: {stream_name}, count: {len(messages)}")
                
            return messages
            
        except Exception as e:
            self.stats['errors'] += 1
            logger.error(f"❌ 读取消息失败: {stream_name}, error: {e}")
            return []
    
    def acknowledge_messages(self, 
                           stream_name: str, 
                           consumer_group: str, 
                           message_ids: List[str]) -> int:
        """确认消息处理完成"""
        try:
            if not message_ids:
                return 0
                
            ack_count = self.redis_client.xack(
                stream_name, 
                consumer_group, 
                *message_ids
            )
            
            logger.debug(f"✅ 消息确认: {stream_name}, acked: {ack_count}/{len(message_ids)}")
            return ack_count
            
        except Exception as e:
            logger.error(f"❌ 消息确认失败: {stream_name}, error: {e}")
            return 0
    
    def get_stream_info(self, stream_name: str) -> Dict[str, Any]:
        """获取Stream信息"""
        try:
            info = self.redis_client.xinfo_stream(stream_name)
            return {
                'name': stream_name,
                'length': info.get('length', 0),
                'groups': info.get('groups', 0),
                'first_entry_id': info.get('first-entry', [None])[0] if info.get('first-entry') else None,
                'last_entry_id': info.get('last-entry', [None])[0] if info.get('last-entry') else None,
            }
        except Exception as e:
            logger.error(f"❌ 获取Stream信息失败: {stream_name}, error: {e}")
            return {}
    
    def get_all_streams_stats(self) -> Dict[str, Any]:
        """获取所有Stream统计信息"""
        stats = {
            'manager_stats': self.stats.copy(),
            'streams': {}
        }
        
        for stream_name, config in self.streams_config.items():
            stream_info = self.get_stream_info(stream_name)
            
            stats['streams'][stream_name] = {
                'stream_info': stream_info,
                'config': config
            }
        
        return stats
    
    def health_check(self) -> bool:
        """健康检查"""
        try:
            # 测试Redis连接
            self.redis_client.ping()
            
            # 检查每个Stream是否可访问
            for stream_name in self.streams_config.keys():
                self.get_stream_info(stream_name)
            
            return True
            
        except Exception as e:
            logger.error(f"❌ 健康检查失败: {e}")
            return False

# 全局单例
_stream_manager = None

def get_stream_manager() -> RedisStreamManager:
    """获取全局Stream管理器实例"""
    global _stream_manager
    if _stream_manager is None:
        _stream_manager = RedisStreamManager()
    return _stream_manager

1.1.2 创建消费者处理器

# 创建消费者文件
touch /Users/brunogao/work/codes/93/release/ljwx-bigscreen/bigscreen/bigScreen/stream_consumers.py

📄 stream_consumers.py 完整代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Redis Stream 消费者处理器
负责从Stream中消费消息并批量处理
"""

import asyncio
import json
import logging
import threading
import time
from typing import List, Dict, Any
from datetime import datetime
from .redis_stream_manager import get_stream_manager, StreamMessage
from .health_data_batch_processor import HealthDataOptimizer
from .models import db
from flask import current_app

logger = logging.getLogger(__name__)

class BaseStreamConsumer:
    """Stream消费者基类"""
    
    def __init__(self, stream_name: str, consumer_name: str, batch_size: int = 200):
        self.stream_name = stream_name
        self.consumer_name = consumer_name
        self.batch_size = batch_size
        self.running = False
        self.stream_manager = get_stream_manager()
        self.stats = {
            'processed_messages': 0,
            'processed_batches': 0,
            'errors': 0,
            'start_time': None
        }
        
    def start(self):
        """启动消费者"""
        if not self.running:
            self.running = True
            self.stats['start_time'] = datetime.now()
            
            # 在单独线程中启动消费循环
            consumer_thread = threading.Thread(
                target=self._consume_loop,
                daemon=True,
                name=f"StreamConsumer-{self.consumer_name}"
            )
            consumer_thread.start()
            
            logger.info(f"🚀 Stream消费者启动: {self.stream_name}:{self.consumer_name}")
    
    def stop(self):
        """停止消费者"""
        self.running = False
        logger.info(f"🛑 Stream消费者停止: {self.stream_name}:{self.consumer_name}")
    
    def _consume_loop(self):
        """消费循环"""
        config = self.stream_manager.streams_config.get(self.stream_name, {})
        consumer_group = config.get('consumer_group', 'default_group')
        
        while self.running:
            try:
                # 从Stream读取消息
                messages = self.stream_manager.read_messages(
                    stream_name=self.stream_name,
                    consumer_group=consumer_group,
                    consumer_name=self.consumer_name,
                    count=self.batch_size,
                    block=1000  # 1秒超时
                )
                
                if messages:
                    # 处理消息批次
                    success = self._process_batch(messages)
                    
                    if success:
                        # 确认消息处理完成
                        message_ids = [msg.stream_id for msg in messages]
                        self.stream_manager.acknowledge_messages(
                            self.stream_name, 
                            consumer_group, 
                            message_ids
                        )
                        
                        # 更新统计
                        self.stats['processed_messages'] += len(messages)
                        self.stats['processed_batches'] += 1
                        
                        logger.debug(f"✅ 批次处理完成: {self.stream_name}, count: {len(messages)}")
                    else:
                        # 处理失败，消息会重新投递
                        self.stats['errors'] += 1
                        logger.error(f"❌ 批次处理失败: {self.stream_name}")
                
            except Exception as e:
                self.stats['errors'] += 1
                logger.error(f"❌ 消费循环异常: {self.stream_name}, error: {e}")
                time.sleep(5)  # 异常时等待5秒再重试
    
    def _process_batch(self, messages: List[StreamMessage]) -> bool:
        """处理消息批次 - 子类实现"""
        raise NotImplementedError("子类必须实现_process_batch方法")
    
    def get_stats(self) -> Dict[str, Any]:
        """获取消费者统计信息"""
        stats = self.stats.copy()
        if stats['start_time']:
            runtime = datetime.now() - stats['start_time']
            stats['runtime_seconds'] = runtime.total_seconds()
            if stats['processed_messages'] > 0:
                stats['messages_per_second'] = stats['processed_messages'] / runtime.total_seconds()
        
        return stats

class HealthDataStreamConsumer(BaseStreamConsumer):
    """健康数据Stream消费者"""
    
    def __init__(self, consumer_name: str = "health_consumer_1"):
        super().__init__('health_data_stream', consumer_name)
        # 复用现有的健康数据优化器
        self.optimizer = HealthDataOptimizer()
    
    def _process_batch(self, messages: List[StreamMessage]) -> bool:
        """处理健康数据批次"""
        try:
            with current_app.app_context():
                # 转换为优化器能理解的格式
                batch_data = []
                
                for message in messages:
                    try:
                        # 从Stream消息中提取数据
                        payload = message.payload
                        
                        # 构造健康数据记录
                        health_record = self._convert_to_health_record(payload)
                        if health_record:
                            batch_data.append(health_record)
                            
                    except Exception as e:
                        logger.error(f"❌ 转换健康数据失败: {message.stream_id}, error: {e}")
                
                # 使用现有优化器批量写入
                if batch_data:
                    self.optimizer._flush_batch(batch_data)
                    return True
                    
            return len(messages) == 0  # 空批次也算成功
            
        except Exception as e:
            logger.error(f"❌ 健康数据批次处理失败: {e}")
            return False
    
    def _convert_to_health_record(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """将Stream消息转换为健康数据记录"""
        try:
            # 从payload中提取数据字段
            data_field = payload.get('data', {})
            
            if isinstance(data_field, list) and len(data_field) > 0:
                # data是数组，取第一个元素
                record = data_field[0]
            elif isinstance(data_field, dict):
                # data是对象
                record = data_field
            else:
                record = payload
            
            # 提取设备SN
            device_sn = (record.get('deviceSn') or 
                        record.get('id') or 
                        payload.get('device_sn'))
            
            if not device_sn:
                return None
            
            # 构造健康数据记录（与现有格式兼容）
            health_record = {
                'device_sn': device_sn,
                'main_data': {
                    'device_sn': device_sn,
                    'heart_rate': record.get('heart_rate'),
                    'blood_oxygen': record.get('blood_oxygen'), 
                    'temperature': record.get('body_temperature'),
                    'pressure_high': record.get('blood_pressure_systolic'),
                    'pressure_low': record.get('blood_pressure_diastolic'),
                    'stress': record.get('stress'),
                    'step': record.get('step'),
                    'distance': record.get('distance'),
                    'calorie': record.get('calorie'),
                    'latitude': record.get('latitude'),
                    'longitude': record.get('longitude'),
                    'altitude': record.get('altitude'),
                    'sleep': record.get('sleepData'),
                    'timestamp': datetime.now(),
                    'upload_method': 'stream_v2'
                }
            }
            
            return health_record
            
        except Exception as e:
            logger.error(f"❌ 健康数据转换失败: {e}")
            return None

class DeviceInfoStreamConsumer(BaseStreamConsumer):
    """设备信息Stream消费者"""
    
    def __init__(self, consumer_name: str = "device_consumer_1"):
        super().__init__('device_info_stream', consumer_name)
    
    def _process_batch(self, messages: List[StreamMessage]) -> bool:
        """处理设备信息批次"""
        try:
            with current_app.app_context():
                # TODO: 实现设备信息批量处理逻辑
                # 可以复用现有的设备信息处理代码
                
                for message in messages:
                    payload = message.payload
                    device_info = payload.get('data', {})
                    
                    # 处理单个设备信息
                    # 这里可以调用现有的device模块处理函数
                    
                logger.info(f"📱 设备信息批次处理: {len(messages)} 条记录")
                return True
                
        except Exception as e:
            logger.error(f"❌ 设备信息批次处理失败: {e}")
            return False

class CommonEventStreamConsumer(BaseStreamConsumer):
    """通用事件Stream消费者"""
    
    def __init__(self, consumer_name: str = "event_consumer_1"):
        super().__init__('common_event_stream', consumer_name)
    
    def _process_batch(self, messages: List[StreamMessage]) -> bool:
        """处理通用事件批次"""
        try:
            with current_app.app_context():
                # TODO: 实现通用事件批量处理逻辑
                # 可以复用现有的告警和事件处理代码
                
                for message in messages:
                    payload = message.payload
                    event_data = payload.get('data', {})
                    
                    # 处理单个事件
                    # 这里可以调用现有的alert模块处理函数
                
                logger.info(f"⚡ 通用事件批次处理: {len(messages)} 条记录")
                return True
                
        except Exception as e:
            logger.error(f"❌ 通用事件批次处理失败: {e}")
            return False

class StreamConsumerManager:
    """Stream消费者管理器"""
    
    def __init__(self):
        self.consumers = {}
        self.running = False
    
    def start_all_consumers(self):
        """启动所有消费者"""
        if not self.running:
            # 启动健康数据消费者
            health_consumer = HealthDataStreamConsumer("health_consumer_1")
            health_consumer.start()
            self.consumers['health_consumer_1'] = health_consumer
            
            # 启动设备信息消费者
            device_consumer = DeviceInfoStreamConsumer("device_consumer_1")  
            device_consumer.start()
            self.consumers['device_consumer_1'] = device_consumer
            
            # 启动通用事件消费者
            event_consumer = CommonEventStreamConsumer("event_consumer_1")
            event_consumer.start()
            self.consumers['event_consumer_1'] = event_consumer
            
            self.running = True
            logger.info("🚀 所有Stream消费者已启动")
    
    def stop_all_consumers(self):
        """停止所有消费者"""
        for consumer in self.consumers.values():
            consumer.stop()
        
        self.consumers.clear()
        self.running = False
        logger.info("🛑 所有Stream消费者已停止")
    
    def get_all_stats(self) -> Dict[str, Any]:
        """获取所有消费者统计信息"""
        stats = {}
        for name, consumer in self.consumers.items():
            stats[name] = consumer.get_stats()
        return stats

# 全局消费者管理器
_consumer_manager = None

def get_consumer_manager() -> StreamConsumerManager:
    """获取全局消费者管理器"""
    global _consumer_manager
    if _consumer_manager is None:
        _consumer_manager = StreamConsumerManager()
    return _consumer_manager

Step 1.2: 生产者接口改造

1.2.1 创建Stream版本的API接口

在 bigScreen.py 中添加新的Stream版本接口：

# 在 bigScreen.py 中添加以下代码

from .redis_stream_manager import get_stream_manager
from .stream_consumers import get_consumer_manager

# 全局Stream管理器
stream_manager = None
consumer_manager = None

def initialize_stream_system():
    """初始化Stream系统"""
    global stream_manager, consumer_manager
    
    try:
        stream_manager = get_stream_manager()
        consumer_manager = get_consumer_manager()
        
        # 启动消费者（仅在验证阶段，不写数据库）
        consumer_manager.start_all_consumers()
        
        logger.info("✅ Stream系统初始化完成")
        return True
    except Exception as e:
        logger.error(f"❌ Stream系统初始化失败: {e}")
        return False

# 在应用启动时调用
with app.app_context():
    initialize_stream_system()

# ============= Stream版本API接口 =============

@app.route("/upload_health_data_v2", methods=['POST'])
@log_api_request('/upload_health_data_v2', 'POST')
def upload_health_data_stream():
    """Redis Stream版本 - 健康数据上传"""
    try:
        health_data = request.get_json()
        
        if not health_data:
            return jsonify({
                "status": "error", 
                "message": "请求体不能为空"
            }), 400
        
        # 提取设备SN用于日志
        data_field = health_data.get('data', {})
        if isinstance(data_field, list) and len(data_field) > 0:
            device_sn = data_field[0].get('deviceSn') or data_field[0].get('id')
        elif isinstance(data_field, dict):
            device_sn = data_field.get('deviceSn') or data_field.get('id')
        else:
            device_sn = "unknown"
        
        # 添加到Stream
        stream_id = stream_manager.add_health_data({
            'data': health_data.get('data'),
            'device_sn': device_sn,
            'message_type': 'health_data',
            'timestamp': int(time.time()),
            'api_version': 'v2'
        })
        
        # 立即响应
        health_logger.info('健康数据Stream上传', extra={
            'device_sn': device_sn,
            'stream_id': stream_id,
            'api_version': 'v2'
        })
        
        return jsonify({
            "status": "accepted",
            "stream_id": stream_id,
            "message": "数据已加入处理队列",
            "processing": "async"
        })
        
    except Exception as e:
        logger.error(f"❌ Stream健康数据上传失败: {e}")
        return jsonify({
            "status": "error",
            "message": f"上传失败: {str(e)}"
        }), 500

@app.route("/upload_device_info_v2", methods=['POST'])
@log_api_request('/upload_device_info_v2', 'POST')
def upload_device_info_stream():
    """Redis Stream版本 - 设备信息上传"""
    try:
        device_info = request.get_json()
        
        if not device_info:
            return jsonify({
                "status": "error", 
                "message": "请求体不能为空"
            }), 400
        
        # 提取设备SN
        device_sn = (device_info.get('SerialNumber') or 
                    device_info.get('deviceSn') or 
                    "unknown")
        
        # 添加到Stream
        stream_id = stream_manager.add_device_info({
            'data': device_info,
            'device_sn': device_sn,
            'message_type': 'device_info',
            'timestamp': int(time.time()),
            'api_version': 'v2'
        })
        
        device_logger.info('设备信息Stream上传', extra={
            'device_sn': device_sn,
            'stream_id': stream_id
        })
        
        return jsonify({
            "status": "accepted",
            "stream_id": stream_id,
            "message": "设备信息已加入处理队列"
        })
        
    except Exception as e:
        logger.error(f"❌ Stream设备信息上传失败: {e}")
        return jsonify({
            "status": "error",
            "message": f"上传失败: {str(e)}"
        }), 500

@app.route("/upload_common_event_v2", methods=['POST'])
@log_api_request('/upload_common_event_v2', 'POST')  
def upload_common_event_stream():
    """Redis Stream版本 - 通用事件上传"""
    try:
        event_data = request.get_json()
        
        if not event_data:
            return jsonify({
                "status": "error",
                "message": "请求体不能为空"
            }), 400
        
        # 提取设备SN
        device_sn = (event_data.get('deviceSn') or 
                    event_data.get('id') or
                    "unknown")
        
        # 添加到Stream  
        stream_id = stream_manager.add_common_event({
            'data': event_data,
            'device_sn': device_sn,
            'message_type': 'common_event',
            'timestamp': int(time.time()),
            'api_version': 'v2'
        })
        
        alert_logger.info('通用事件Stream上传', extra={
            'device_sn': device_sn,
            'stream_id': stream_id,
            'event_type': event_data.get('eventType', 'unknown')
        })
        
        return jsonify({
            "status": "accepted", 
            "stream_id": stream_id,
            "message": "事件已加入处理队列"
        })
        
    except Exception as e:
        logger.error(f"❌ Stream事件上传失败: {e}")
        return jsonify({
            "status": "error",
            "message": f"上传失败: {str(e)}"
        }), 500

# ============= Stream监控接口 =============

@app.route("/api/stream_stats", methods=['GET'])
def get_stream_stats():
    """获取Stream统计信息"""
    try:
        if stream_manager is None:
            return jsonify({"error": "Stream系统未初始化"}), 503
            
        stats = stream_manager.get_all_streams_stats()
        consumer_stats = consumer_manager.get_all_stats() if consumer_manager else {}
        
        return jsonify({
            "stream_stats": stats,
            "consumer_stats": consumer_stats,
            "timestamp": int(time.time())
        })
        
    except Exception as e:
        logger.error(f"❌ 获取Stream统计失败: {e}")
        return jsonify({"error": str(e)}), 500

@app.route("/api/stream_health", methods=['GET'])
def check_stream_health():
    """Stream健康检查"""
    try:
        if stream_manager is None:
            return jsonify({
                "healthy": False,
                "error": "Stream系统未初始化"
            }), 503
        
        healthy = stream_manager.health_check()
        
        return jsonify({
            "healthy": healthy,
            "timestamp": int(time.time()),
            "streams": list(stream_manager.streams_config.keys())
        })
        
    except Exception as e:
        logger.error(f"❌ Stream健康检查失败: {e}")
        return jsonify({
            "healthy": False,
            "error": str(e)
        }), 500

1.2.2 部署验证

# 1. 启动应用
cd /Users/brunogao/work/codes/93/release/ljwx-bigscreen/bigscreen
python run_bigscreen.py

# 2. 验证Stream系统
curl http://localhost:5225/api/stream_health

# 3. 测试Stream接口
curl -X POST http://localhost:5225/upload_health_data_v2 \
  -H "Content-Type: application/json" \
  -d '{"data": {"deviceSn": "TEST001", "heart_rate": 80}}'

Step 1.3: 双写验证机制

1.3.1 实现双写对比验证

创建验证工具来对比新旧系统的数据一致性：

# 创建验证工具
touch /Users/brunogao/work/codes/93/release/ljwx-bigscreen/bigscreen/stream_validation_tool.py

📄 stream_validation_tool.py 验证工具代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Stream数据验证工具
对比新旧系统的数据一致性
"""

import time
import json
import requests
import logging
import threading
from datetime import datetime, timedelta
from typing import Dict, List, Any
from concurrent.futures import ThreadPoolExecutor, as_completed

logger = logging.getLogger(__name__)

class StreamValidationTool:
    """Stream数据验证工具"""
    
    def __init__(self, base_url: str = "http://localhost:5225"):
        self.base_url = base_url
        self.validation_results = {
            'health_data': {'total': 0, 'success': 0, 'failed': 0, 'errors': []},
            'device_info': {'total': 0, 'success': 0, 'failed': 0, 'errors': []},
            'common_event': {'total': 0, 'success': 0, 'failed': 0, 'errors': []}
        }
        
    def validate_health_data_consistency(self, test_data: List[Dict], duration_minutes: int = 10):
        """验证健康数据一致性"""
        print(f"🚀 开始健康数据一致性验证 - 持续 {duration_minutes} 分钟")
        
        end_time = datetime.now() + timedelta(minutes=duration_minutes)
        
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = []
            
            while datetime.now() < end_time:
                for data in test_data:
                    # 提交双写验证任务
                    future = executor.submit(self._validate_single_health_data, data)
                    futures.append(future)
                    
                    time.sleep(0.1)  # 控制发送频率
                
                # 处理已完成的任务
                completed_futures = [f for f in futures if f.done()]
                for future in completed_futures:
                    try:
                        result = future.result()
                        self._record_validation_result('health_data', result)
                    except Exception as e:
                        logger.error(f"验证任务异常: {e}")
                    
                    futures.remove(future)
        
        # 等待剩余任务完成
        for future in as_completed(futures, timeout=60):
            try:
                result = future.result()
                self._record_validation_result('health_data', result)
            except Exception as e:
                logger.error(f"最终验证任务异常: {e}")
        
        self._print_validation_summary('health_data')
    
    def _validate_single_health_data(self, test_data: Dict) -> Dict[str, Any]:
        """验证单条健康数据"""
        try:
            # 1. 发送到旧版接口
            old_response = requests.post(
                f"{self.base_url}/upload_health_data",
                json=test_data,
                timeout=10
            )
            
            # 2. 发送到新版Stream接口
            new_response = requests.post(
                f"{self.base_url}/upload_health_data_v2", 
                json=test_data,
                timeout=10
            )
            
            # 3. 对比响应
            result = {
                'timestamp': datetime.now().isoformat(),
                'test_data_id': test_data.get('data', {}).get('deviceSn', 'unknown'),
                'old_status': old_response.status_code,
                'new_status': new_response.status_code,
                'old_response': old_response.json() if old_response.status_code == 200 else None,
                'new_response': new_response.json() if new_response.status_code == 200 else None,
                'success': old_response.status_code == 200 and new_response.status_code == 200,
                'consistent': self._compare_responses(old_response, new_response)
            }
            
            return result
            
        except Exception as e:
            return {
                'timestamp': datetime.now().isoformat(),
                'test_data_id': test_data.get('data', {}).get('deviceSn', 'unknown'),
                'success': False,
                'error': str(e)
            }
    
    def _compare_responses(self, old_response, new_response) -> bool:
        """对比新旧接口响应"""
        if old_response.status_code != new_response.status_code:
            return False
            
        # 检查是否都成功
        if old_response.status_code == 200 and new_response.status_code == 200:
            # 新接口返回异步响应，旧接口返回同步响应，这是预期的不同
            old_data = old_response.json()
            new_data = new_response.json()
            
            # 检查关键字段
            old_success = old_data.get('status') == 'success'
            new_success = new_data.get('status') == 'accepted'
            
            return old_success and new_success
        
        return True
    
    def _record_validation_result(self, data_type: str, result: Dict[str, Any]):
        """记录验证结果"""
        stats = self.validation_results[data_type]
        stats['total'] += 1
        
        if result.get('success', False):
            stats['success'] += 1
        else:
            stats['failed'] += 1
            stats['errors'].append(result)
    
    def _print_validation_summary(self, data_type: str):
        """打印验证摘要"""
        stats = self.validation_results[data_type]
        
        print(f"\n📊 {data_type} 验证结果摘要:")
        print(f"   总请求数: {stats['total']}")
        print(f"   成功数: {stats['success']}")
        print(f"   失败数: {stats['failed']}")
        print(f"   成功率: {stats['success']/stats['total']*100:.2f}%" if stats['total'] > 0 else "   成功率: 0%")
        
        if stats['errors']:
            print(f"   错误示例:")
            for error in stats['errors'][:3]:  # 只显示前3个错误
                print(f"     - {error.get('error', 'Unknown error')}")
    
    def generate_test_health_data(self, device_count: int = 10) -> List[Dict]:
        """生成测试健康数据"""
        test_data = []
        
        for i in range(device_count):
            data = {
                "data": {
                    "deviceSn": f"STREAM_TEST_{i:03d}",
                    "heart_rate": 70 + (i % 30),
                    "blood_oxygen": 95 + (i % 5),
                    "body_temperature": 36.5 + (i % 2),
                    "step": 1000 + (i * 100),
                    "timestamp": int(time.time())
                }
            }
            test_data.append(data)
        
        return test_data
    
    def run_comprehensive_validation(self):
        """运行综合验证"""
        print("🎯 开始Stream系统综合验证")
        
        # 生成测试数据
        test_health_data = self.generate_test_health_data(20)
        
        # 验证健康数据一致性
        self.validate_health_data_consistency(test_health_data, duration_minutes=5)
        
        # TODO: 添加设备信息和通用事件验证
        
        print("\n✅ 综合验证完成")
        return self.validation_results

def main():
    """验证工具主函数"""
    validator = StreamValidationTool()
    results = validator.run_comprehensive_validation()
    
    # 保存验证结果
    with open(f'stream_validation_results_{int(time.time())}.json', 'w') as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
    
    print("📄 验证结果已保存到文件")

if __name__ == "__main__":
    main()

---

📋 Phase 2: 灰度测试 (Week 3)

Step 2.1: 流量切换配置

2.1.1 创建流量分流器

# 在 bigScreen.py 中添加流量分流逻辑

import random

# 流量分流配置
STREAM_TRAFFIC_RATIO = 0.1  # 10% 流量切到Stream

def should_use_stream_api() -> bool:
    """判断是否使用Stream API"""
    return random.random() < STREAM_TRAFFIC_RATIO

@app.route("/upload_health_data", methods=['POST'])  
@log_api_request('/upload_health_data','POST')
def handle_health_data():
    """智能路由版本 - 自动分流到Stream或传统处理"""
    
    # 流量分流判断
    use_stream = should_use_stream_api()
    
    if use_stream:
        # 路由到Stream处理
        return upload_health_data_stream()
    else:
        # 保持原有处理逻辑
        health_data = request.get_json()
        # ... 原有代码保持不变
        result = optimized_upload_health_data(health_data)
        return result

2.1.2 动态调整流量比例

# 添加流量比例动态调整接口
@app.route("/api/stream_traffic_ratio", methods=['GET', 'POST'])
def manage_stream_traffic_ratio():
    """管理Stream流量比例"""
    global STREAM_TRAFFIC_RATIO
    
    if request.method == 'GET':
        return jsonify({
            "current_ratio": STREAM_TRAFFIC_RATIO,
            "description": f"{STREAM_TRAFFIC_RATIO*100:.1f}% 流量使用Stream"
        })
    
    elif request.method == 'POST':
        data = request.get_json()
        new_ratio = data.get('ratio', STREAM_TRAFFIC_RATIO)
        
        # 安全检查
        if 0 <= new_ratio <= 1:
            old_ratio = STREAM_TRAFFIC_RATIO
            STREAM_TRAFFIC_RATIO = new_ratio
            
            logger.info(f"🔄 Stream流量比例调整: {old_ratio*100:.1f}% -> {new_ratio*100:.1f}%")
            
            return jsonify({
                "success": True,
                "old_ratio": old_ratio,
                "new_ratio": new_ratio,
                "message": f"流量比例已调整为 {new_ratio*100:.1f}%"
            })
        else:
            return jsonify({
                "error": "比例必须在 0-1 之间"
            }), 400

Step 2.2: 监控和告警

2.2.1 关键指标监控

# 添加详细监控指标收集
class StreamMetrics:
    def __init__(self):
        self.metrics = {
            'requests': {'stream': 0, 'traditional': 0},
            'response_times': {'stream': [], 'traditional': []},
            'errors': {'stream': 0, 'traditional': 0},
            'throughput': {'stream': 0, 'traditional': 0}
        }
        self.last_reset = time.time()
    
    def record_request(self, api_type: str, response_time: float, success: bool):
        """记录请求指标"""
        self.metrics['requests'][api_type] += 1
        self.metrics['response_times'][api_type].append(response_time)
        
        if not success:
            self.metrics['errors'][api_type] += 1
    
    def get_summary(self) -> Dict[str, Any]:
        """获取指标摘要"""
        now = time.time()
        duration = now - self.last_reset
        
        summary = {}
        
        for api_type in ['stream', 'traditional']:
            requests = self.metrics['requests'][api_type]
            response_times = self.metrics['response_times'][api_type]
            errors = self.metrics['errors'][api_type]
            
            summary[api_type] = {
                'requests': requests,
                'qps': requests / duration if duration > 0 else 0,
                'avg_response_time': sum(response_times) / len(response_times) if response_times else 0,
                'p95_response_time': sorted(response_times)[int(len(response_times) * 0.95)] if response_times else 0,
                'error_rate': errors / requests * 100 if requests > 0 else 0,
                'errors': errors
            }
        
        return {
            'summary': summary,
            'duration_seconds': duration,
            'timestamp': now
        }

# 全局指标收集器
stream_metrics = StreamMetrics()

@app.route("/api/stream_metrics", methods=['GET'])
def get_stream_metrics():
    """获取Stream性能指标"""
    return jsonify(stream_metrics.get_summary())

Step 2.3: 灰度测试执行

2.3.1 测试脚本

#!/bin/bash
# 灰度测试执行脚本

echo "🎯 开始Stream灰度测试"

# 1. 设置10%流量
curl -X POST http://localhost:5225/api/stream_traffic_ratio \
  -H "Content-Type: application/json" \
  -d '{"ratio": 0.1}'

echo "✅ 流量比例设置为10%"

# 2. 运行负载测试
python3 stream_validation_tool.py

# 3. 监控30分钟
for i in {1..30}; do
  echo "📊 监控第 $i 分钟..."
  curl -s http://localhost:5225/api/stream_metrics | jq '.summary'
  sleep 60
done

echo "📈 灰度测试完成"

---

📋 Phase 3: 全量切换 (Week 4)

Step 3.1: 切换准备

3.1.1 数据一致性验证

# 数据库一致性检查工具
def verify_database_consistency():
    """验证数据库数据一致性"""
    
    # 检查最近1小时的数据
    one_hour_ago = datetime.now() - timedelta(hours=1)
    
    # 查询传统方式和Stream方式处理的数据
    traditional_count = db.session.query(UserHealthData).filter(
        UserHealthData.upload_method == 'optimized',
        UserHealthData.create_time >= one_hour_ago
    ).count()
    
    stream_count = db.session.query(UserHealthData).filter(
        UserHealthData.upload_method == 'stream_v2',
        UserHealthData.create_time >= one_hour_ago
    ).count()
    
    total_expected = traditional_count + stream_count
    
    print(f"📊 数据一致性检查:")
    print(f"   传统方式: {traditional_count} 条")
    print(f"   Stream方式: {stream_count} 条") 
    print(f"   总计: {total_expected} 条")
    
    # 检查重复数据
    duplicates = db.session.query(UserHealthData).filter(
        UserHealthData.create_time >= one_hour_ago
    ).group_by(
        UserHealthData.device_sn,
        UserHealthData.timestamp
    ).having(
        func.count(UserHealthData.id) > 1
    ).count()
    
    print(f"   重复数据: {duplicates} 条")
    
    return {
        'traditional_count': traditional_count,
        'stream_count': stream_count,
        'total_count': total_expected,
        'duplicates': duplicates,
        'consistency_ok': duplicates == 0
    }

3.1.2 回滚预案准备

# 回滚预案实现
@app.route("/api/emergency_rollback", methods=['POST'])
def emergency_rollback():
    """紧急回滚到传统处理方式"""
    global STREAM_TRAFFIC_RATIO
    
    try:
        # 1. 立即停止Stream流量
        STREAM_TRAFFIC_RATIO = 0.0
        
        # 2. 停止Stream消费者
        if consumer_manager:
            consumer_manager.stop_all_consumers()
        
        # 3. 记录回滚日志
        rollback_time = datetime.now()
        logger.critical(f"🚨 紧急回滚执行: {rollback_time}")
        
        # 4. 验证回滚效果
        time.sleep(5)  # 等待5秒
        metrics = stream_metrics.get_summary()
        
        return jsonify({
            "success": True,
            "rollback_time": rollback_time.isoformat(),
            "traffic_ratio": STREAM_TRAFFIC_RATIO,
            "consumers_stopped": True,
            "current_metrics": metrics,
            "message": "已紧急回滚到传统处理方式"
        })
        
    except Exception as e:
        logger.error(f"❌ 紧急回滚失败: {e}")
        return jsonify({
            "success": False,
            "error": str(e)
        }), 500

Step 3.2: 全量切换执行

3.2.1 切换执行脚本

#!/bin/bash
# 全量切换执行脚本

set -e  # 遇到错误立即停止

echo "🚀 开始Stream全量切换"

# 1. 预检查
echo "🔍 执行预检查..."
python3 -c "
import requests
import sys

# 检查系统健康
health = requests.get('http://localhost:5225/api/stream_health').json()
if not health.get('healthy'):
    print('❌ Stream系统不健康，停止切换')
    sys.exit(1)

print('✅ Stream系统健康检查通过')
"

# 2. 数据一致性验证
echo "📊 验证数据一致性..."
python3 -c "
# 调用数据一致性验证函数
# verify_database_consistency()
"

# 3. 分阶段切换
echo "📈 分阶段流量切换..."

# 30% -> 60% -> 100%
for ratio in 0.3 0.6 1.0; do
  echo "🔄 设置流量比例为 ${ratio}"
  curl -X POST http://localhost:5225/api/stream_traffic_ratio \
    -H "Content-Type: application/json" \
    -d "{\"ratio\": ${ratio}}"
  
  # 观察5分钟
  echo "⏱️  观察 5 分钟..."
  for i in {1..5}; do
    metrics=$(curl -s http://localhost:5225/api/stream_metrics | jq -r '.summary.stream.error_rate')
    echo "   错误率: ${metrics}%"
    
    # 检查错误率阈值
    if (( $(echo "${metrics} > 1.0" | bc -l) )); then
      echo "❌ 错误率过高，执行回滚"
      curl -X POST http://localhost:5225/api/emergency_rollback
      exit 1
    fi
    
    sleep 60
  done
done

# 4. 关闭传统处理器
echo "🔄 关闭传统批处理器..."
# 这里可以添加关闭传统HealthDataOptimizer的逻辑

echo "✅ Stream全量切换完成！"
echo "📊 最终性能统计:"
curl -s http://localhost:5225/api/stream_metrics | jq '.summary'

Step 3.3: 切换后优化

3.3.1 性能调优

# 性能自动调优
class StreamPerformanceOptimizer:
    def __init__(self):
        self.optimization_history = []
    
    def auto_optimize(self):
        """自动性能优化"""
        metrics = stream_metrics.get_summary()
        
        # 根据QPS调整消费者数量
        current_qps = metrics['summary']['stream']['qps']
        
        if current_qps > 2000:
            # 高负载，增加消费者
            self._scale_up_consumers()
        elif current_qps < 500:
            # 低负载，减少消费者
            self._scale_down_consumers()
        
        # 根据延迟调整批次大小
        avg_response_time = metrics['summary']['stream']['avg_response_time']
        
        if avg_response_time > 500:  # 500ms
            # 响应时间过长，减少批次大小
            self._reduce_batch_size()
        elif avg_response_time < 100:  # 100ms
            # 响应时间很快，可以增加批次大小
            self._increase_batch_size()
    
    def _scale_up_consumers(self):
        """扩展消费者"""
        # 添加更多消费者实例
        pass
    
    def _scale_down_consumers(self):
        """减少消费者"""
        # 减少消费者实例
        pass
    
    def _reduce_batch_size(self):
        """减少批次大小"""
        pass
    
    def _increase_batch_size(self):
        """增加批次大小"""
        pass

# 启动性能优化器
performance_optimizer = StreamPerformanceOptimizer()

# 定期优化任务
def periodic_optimization():
    while True:
        try:
            performance_optimizer.auto_optimize()
        except Exception as e:
            logger.error(f"性能优化失败: {e}")
        
        time.sleep(300)  # 每5分钟优化一次

# 启动优化线程
threading.Thread(target=periodic_optimization, daemon=True).start()

---

🎯 总体切换时间表

Week 1-2: 基础设施 + 并行验证

• Day 1-2: 部署Redis Stream管理器和消费者
• Day 3-4: 创建Stream版本API接口
• Day 5-7: 实施双写验证机制
• Day 8-10: 并行运行验证，确保数据一致性
• Day 11-14: 性能测试和问题修复

Week 3: 灰度测试

• Day 15: 10% 流量切换
• Day 16-17: 监控关键指标，调整参数
• Day 18: 30% 流量切换
• Day 19-20: 性能对比和稳定性验证
• Day 21: 50% 流量切换，全面测试

Week 4: 全量切换

• Day 22: 最终数据一致性验证
• Day 23: 80% 流量切换
• Day 24: 100% 流量切换
• Day 25-27: 性能监控和优化调整
• Day 28: 关闭传统处理器，切换完成

---

⚠️ 关键风险点和应对策略

🔴 高风险操作点

1. Redis连接中断

   • 监控: 实时监控Redis连接状态
   • 应对: 自动重连 + 熔断降级到内存队列

2. 消费者处理延迟

   • 监控: Stream消息堆积数量
   • 应对: 动态扩展消费者数量

3. 数据不一致

   • 监控: 定期对比新旧系统数据库记录
   • 应对: 立即回滚 + 数据修复脚本

🛡️ 安全保障措施

1. 实时监控大屏
2. 自动告警系统
3. 一键回滚机制
4. 数据备份机制
5. 分阶段切换策略

---

📞 应急联系和支持

关键联系人

• 系统负责人: [具体联系方式]
• DBA支持: [数据库支持联系方式]
• 运维支持: [运维团队联系方式]

应急处理流程

1. 发现问题 → 立即记录和通报
2. 评估影响 → 判断是否需要回滚
3. 执行回滚 → 使用一键回滚接口
4. 问题修复 → 离线修复后重新上线
5. 复盘总结 → 优化流程避免重复

---

通过这个详细的迁移指南，您可以安全、平稳地完成从传统批处理到Redis Stream的架构升级，实现性能的显著提升。