import logging
import threading
import time
from concurrent.futures import ThreadPoolExecutor

# 配置基礎(chǔ)日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(threadName)s] %(levelname)s: %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def worker_task(task_id):
    """模擬工作任務(wù)，產(chǎn)生大量日志"""
    for i in range(100):
        # 模擬復(fù)雜的日志消息
        message = f"Task {task_id} processing item {i} with data: " + "x" * 50
        logger.info(message)
        
        # 模擬一些處理時(shí)間
        time.sleep(0.001)
        
        # 記錄處理結(jié)果
        logger.info(f"Task {task_id} completed item {i} successfully")

def reproduce_log_corruption():
    """重現(xiàn)日志錯(cuò)亂問(wèn)題"""
    print("開(kāi)始重現(xiàn)多線程日志錯(cuò)亂問(wèn)題...")
    
    # 使用線程池執(zhí)行多個(gè)任務(wù)
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(worker_task, i) for i in range(5)]
        
        # 等待所有任務(wù)完成
        for future in futures:
            future.result()
    
    print("任務(wù)執(zhí)行完成，請(qǐng)檢查 app.log 文件中的日志錯(cuò)亂情況")

if __name__ == "__main__":
    reproduce_log_corruption()

import logging
import threading
import inspect

class ThreadSafeAnalyzer:
    """分析logging模塊的線程安全機(jī)制"""
    
    def __init__(self):
        self.logger = logging.getLogger('analyzer')
        self.handler = logging.StreamHandler()
        self.logger.addHandler(self.handler)
    
    def analyze_handler_locks(self):
        """分析Handler的鎖機(jī)制"""
        print("=== Handler鎖機(jī)制分析 ===")
        
        # 檢查Handler是否有鎖
        if hasattr(self.handler, 'lock'):
            print(f"Handler鎖類型: {type(self.handler.lock)}")
            print(f"鎖對(duì)象: {self.handler.lock}")
        else:
            print("Handler沒(méi)有鎖機(jī)制")
        
        # 查看Handler的emit方法源碼結(jié)構(gòu)
        emit_source = inspect.getsource(self.handler.emit)
        print(f"emit方法長(zhǎng)度: {len(emit_source.split('\\n'))} 行")
        
    def analyze_logger_locks(self):
        """分析Logger的鎖機(jī)制"""
        print("\\n=== Logger鎖機(jī)制分析 ===")
        
        # Logger級(jí)別的鎖
        if hasattr(logging, '_lock'):
            print(f"全局鎖: {logging._lock}")
        
        # 檢查L(zhǎng)ogger的線程安全方法
        thread_safe_methods = ['_log', 'handle', 'callHandlers']
        for method in thread_safe_methods:
            if hasattr(self.logger, method):
                print(f"線程安全方法: {method}")

def custom_handler_with_detailed_locking():
    """自定義Handler展示詳細(xì)的鎖機(jī)制"""
    
    class DetailedLockingHandler(logging.StreamHandler):
        def __init__(self, stream=None):
            super().__init__(stream)
            self.emit_count = 0
            self.lock_wait_time = 0
        
        def emit(self, record):
            """重寫(xiě)emit方法，添加詳細(xì)的鎖分析"""
            import time
            
            # 記錄嘗試獲取鎖的時(shí)間
            start_time = time.time()
            
            # 獲取鎖（這里會(huì)調(diào)用父類的acquire方法）
            self.acquire()
            try:
                # 記錄獲取鎖后的時(shí)間
                lock_acquired_time = time.time()
                self.lock_wait_time += (lock_acquired_time - start_time)
                self.emit_count += 1
                
                # 模擬格式化和寫(xiě)入過(guò)程
                if self.stream:
                    msg = self.format(record)
                    # 添加鎖信息到日志中
                    enhanced_msg = f"[EMIT#{self.emit_count}|WAIT:{(lock_acquired_time - start_time)*1000:.2f}ms] {msg}"
                    
                    self.stream.write(enhanced_msg + '\\n')
                    self.flush()
                    
            finally:
                self.release()
        
        def get_stats(self):
            """獲取鎖統(tǒng)計(jì)信息"""
            return {
                'total_emits': self.emit_count,
                'total_wait_time': self.lock_wait_time,
                'avg_wait_time': self.lock_wait_time / max(1, self.emit_count)
            }
    
    return DetailedLockingHandler()

# 使用示例
if __name__ == "__main__":
    analyzer = ThreadSafeAnalyzer()
    analyzer.analyze_handler_locks()
    analyzer.analyze_logger_locks()

import logging
import threading
import time
from typing import List, Dict, Any

class RaceConditionDemo:
    """演示競(jìng)態(tài)條件的具體場(chǎng)景"""
    
    def __init__(self):
        self.race_conditions: List[Dict[str, Any]] = []
        self.lock = threading.Lock()
    
    def simulate_emit_race_condition(self):
        """模擬emit方法中的競(jìng)態(tài)條件"""
        
        class RacyHandler(logging.Handler):
            def __init__(self, demo_instance):
                super().__init__()
                self.demo = demo_instance
                self.step_counter = 0
            
            def emit(self, record):
                """模擬有競(jìng)態(tài)條件的emit實(shí)現(xiàn)"""
                thread_id = threading.current_thread().ident
                
                # 步驟1: 格式化消息（可能被中斷）
                self.demo.log_step(thread_id, "開(kāi)始格式化消息")
                formatted_msg = self.format(record)
                
                # 模擬格式化過(guò)程中的延遲
                time.sleep(0.001)
                
                # 步驟2: 準(zhǔn)備寫(xiě)入（關(guān)鍵競(jìng)態(tài)點(diǎn)）
                self.demo.log_step(thread_id, "準(zhǔn)備寫(xiě)入文件")
                
                # 步驟3: 實(shí)際寫(xiě)入操作
                self.demo.log_step(thread_id, f"寫(xiě)入消息: {formatted_msg[:50]}...")
                
                # 模擬寫(xiě)入過(guò)程的非原子性
                parts = [formatted_msg[i:i+10] for i in range(0, len(formatted_msg), 10)]
                for i, part in enumerate(parts):
                    print(f"[Thread-{thread_id}] Part {i}: {part}")
                    time.sleep(0.0001)  # 模擬寫(xiě)入延遲
                
                self.demo.log_step(thread_id, "寫(xiě)入完成")
        
        return RacyHandler(self)
    
    def log_step(self, thread_id: int, step: str):
        """記錄執(zhí)行步驟"""
        with self.lock:
            self.race_conditions.append({
                'thread_id': thread_id,
                'timestamp': time.time(),
                'step': step
            })
    
    def analyze_race_conditions(self):
        """分析競(jìng)態(tài)條件"""
        print("\\n=== 競(jìng)態(tài)條件分析 ===")
        
        # 按時(shí)間排序
        sorted_steps = sorted(self.race_conditions, key=lambda x: x['timestamp'])
        
        # 分析交錯(cuò)執(zhí)行
        thread_states = {}
        for step in sorted_steps:
            thread_id = step['thread_id']
            if thread_id not in thread_states:
                thread_states[thread_id] = []
            thread_states[thread_id].append(step['step'])
        
        # 檢測(cè)競(jìng)態(tài)模式
        race_patterns = []
        for i in range(len(sorted_steps) - 1):
            current = sorted_steps[i]
            next_step = sorted_steps[i + 1]
            
            if (current['thread_id'] != next_step['thread_id'] and 
                '寫(xiě)入' in current['step'] and '寫(xiě)入' in next_step['step']):
                race_patterns.append({
                    'pattern': 'concurrent_write',
                    'threads': [current['thread_id'], next_step['thread_id']],
                    'time_gap': next_step['timestamp'] - current['timestamp']
                })
        
        return race_patterns

def demonstrate_formatter_race_condition():
    """演示Formatter中的競(jìng)態(tài)條件"""
    
    class StatefulFormatter(logging.Formatter):
        """有狀態(tài)的格式化器，容易產(chǎn)生競(jìng)態(tài)條件"""
        
        def __init__(self):
            super().__init__()
            self.counter = 0
            self.thread_info = {}
        
        def format(self, record):
            """非線程安全的格式化方法"""
            thread_id = threading.current_thread().ident
            
            # 競(jìng)態(tài)條件1: 共享計(jì)數(shù)器
            self.counter += 1
            current_count = self.counter
            
            # 模擬格式化延遲
            time.sleep(0.001)
            
            # 競(jìng)態(tài)條件2: 共享字典
            self.thread_info[thread_id] = {
                'last_message': record.getMessage(),
                'count': current_count
            }
            
            # 構(gòu)建格式化消息
            formatted = f"[{current_count:04d}] {record.levelname}: {record.getMessage()}"
            
            return formatted
    
    # 測(cè)試有狀態(tài)格式化器的競(jìng)態(tài)問(wèn)題
    logger = logging.getLogger('race_test')
    handler = logging.StreamHandler()
    handler.setFormatter(StatefulFormatter())
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    
    def worker(worker_id):
        for i in range(10):
            logger.info(f"Worker {worker_id} message {i}")
    
    # 啟動(dòng)多個(gè)線程
    threads = []
    for i in range(5):
        t = threading.Thread(target=worker, args=(i,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()

if __name__ == "__main__":
    # 演示競(jìng)態(tài)條件
    demo = RaceConditionDemo()
    handler = demo.simulate_emit_race_condition()
    
    logger = logging.getLogger('race_demo')
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    
    # 多線程測(cè)試
    def test_worker(worker_id):
        for i in range(3):
            logger.info(f"Worker {worker_id} executing task {i}")
    
    threads = []
    for i in range(3):
        t = threading.Thread(target=test_worker, args=(i,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    # 分析結(jié)果
    patterns = demo.analyze_race_conditions()
    print(f"檢測(cè)到 {len(patterns)} 個(gè)競(jìng)態(tài)模式")

import logging
import logging.handlers
import queue
import threading
import time
from concurrent.futures import ThreadPoolExecutor

class ThreadSafeLoggingSystem:
    """線程安全的日志系統(tǒng)實(shí)現(xiàn)"""
    
    def __init__(self, log_file='safe_app.log', max_queue_size=1000):
        self.log_queue = queue.Queue(maxsize=max_queue_size)
        self.setup_logging(log_file)
        self.start_log_listener()
    
    def setup_logging(self, log_file):
        """設(shè)置日志配置"""
        # 創(chuàng)建隊(duì)列處理器
        queue_handler = logging.handlers.QueueHandler(self.log_queue)
        
        # 配置根日志器
        root_logger = logging.getLogger()
        root_logger.setLevel(logging.INFO)
        root_logger.addHandler(queue_handler)
        
        # 創(chuàng)建監(jiān)聽(tīng)器處理器
        file_handler = logging.FileHandler(log_file)
        console_handler = logging.StreamHandler()
        
        # 設(shè)置格式化器
        formatter = logging.Formatter(
            '%(asctime)s [%(threadName)-12s] %(levelname)-8s: %(message)s'
        )
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        
        # 創(chuàng)建隊(duì)列監(jiān)聽(tīng)器
        self.queue_listener = logging.handlers.QueueListener(
            self.log_queue,
            file_handler,
            console_handler,
            respect_handler_level=True
        )
    
    def start_log_listener(self):
        """啟動(dòng)日志監(jiān)聽(tīng)器"""
        self.queue_listener.start()
        print("日志監(jiān)聽(tīng)器已啟動(dòng)")
    
    def stop_log_listener(self):
        """停止日志監(jiān)聽(tīng)器"""
        self.queue_listener.stop()
        print("日志監(jiān)聽(tīng)器已停止")
    
    def get_logger(self, name):
        """獲取日志器"""
        return logging.getLogger(name)

class AdvancedQueueHandler(logging.handlers.QueueHandler):
    """增強(qiáng)的隊(duì)列處理器"""
    
    def __init__(self, queue_obj, max_retries=3, retry_delay=0.1):
        super().__init__(queue_obj)
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self.dropped_logs = 0
        self.total_logs = 0
    
    def emit(self, record):
        """重寫(xiě)emit方法，添加重試機(jī)制"""
        self.total_logs += 1
        
        for attempt in range(self.max_retries):
            try:
                self.enqueue(record)
                return
            except queue.Full:
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay)
                    continue
                else:
                    self.dropped_logs += 1
                    # 可以選擇寫(xiě)入到備用日志或者直接丟棄
                    self.handle_dropped_log(record)
                    break
            except Exception as e:
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay)
                    continue
                else:
                    self.handleError(record)
                    break
    
    def handle_dropped_log(self, record):
        """處理被丟棄的日志"""
        # 可以實(shí)現(xiàn)備用策略，比如寫(xiě)入到緊急日志文件
        emergency_msg = f"DROPPED LOG: {record.getMessage()}"
        print(f"WARNING: {emergency_msg}")
    
    def get_stats(self):
        """獲取統(tǒng)計(jì)信息"""
        return {
            'total_logs': self.total_logs,
            'dropped_logs': self.dropped_logs,
            'success_rate': (self.total_logs - self.dropped_logs) / max(1, self.total_logs)
        }

def test_thread_safe_logging():
    """測(cè)試線程安全的日志系統(tǒng)"""
    
    # 初始化線程安全日志系統(tǒng)
    log_system = ThreadSafeLoggingSystem()
    logger = log_system.get_logger('test_app')
    
    def intensive_logging_task(task_id, num_logs=100):
        """密集日志記錄任務(wù)"""
        for i in range(num_logs):
            logger.info(f"Task {task_id} - Processing item {i}")
            logger.debug(f"Task {task_id} - Debug info for item {i}")
            
            if i % 10 == 0:
                logger.warning(f"Task {task_id} - Checkpoint at item {i}")
            
            # 模擬一些處理時(shí)間
            time.sleep(0.001)
        
        logger.info(f"Task {task_id} completed successfully")
    
    print("開(kāi)始線程安全日志測(cè)試...")
    start_time = time.time()
    
    # 使用線程池執(zhí)行多個(gè)任務(wù)
    with ThreadPoolExecutor(max_workers=20) as executor:
        futures = [
            executor.submit(intensive_logging_task, i, 50) 
            for i in range(10)
        ]
        
        # 等待所有任務(wù)完成
        for future in futures:
            future.result()
    
    end_time = time.time()
    print(f"測(cè)試完成，耗時(shí): {end_time - start_time:.2f} 秒")
    
    # 停止日志系統(tǒng)
    log_system.stop_log_listener()
    
    return log_system

if __name__ == "__main__":
    test_thread_safe_logging()

import logging
import threading
import time
import contextlib
from typing import Optional, Dict, Any

class SynchronizedHandler(logging.Handler):
    """完全同步的日志處理器"""
    
    def __init__(self, target_handler: logging.Handler):
        super().__init__()
        self.target_handler = target_handler
        self.emit_lock = threading.RLock()  # 使用可重入鎖
        self.format_lock = threading.RLock()
        
        # 統(tǒng)計(jì)信息
        self.stats = {
            'total_emits': 0,
            'lock_wait_time': 0.0,
            'max_wait_time': 0.0,
            'concurrent_attempts': 0
        }
    
    def emit(self, record):
        """完全同步的emit實(shí)現(xiàn)"""
        start_wait = time.time()
        
        with self.emit_lock:
            wait_time = time.time() - start_wait
            self.stats['lock_wait_time'] += wait_time
            self.stats['max_wait_time'] = max(self.stats['max_wait_time'], wait_time)
            self.stats['total_emits'] += 1
            
            try:
                # 同步格式化
                with self.format_lock:
                    if self.formatter:
                        record.message = record.getMessage()
                        formatted = self.formatter.format(record)
                    else:
                        formatted = record.getMessage()
                
                # 同步寫(xiě)入
                self.target_handler.emit(record)
                
            except Exception as e:
                self.handleError(record)
    
    def get_performance_stats(self) -> Dict[str, Any]:
        """獲取性能統(tǒng)計(jì)"""
        total_emits = max(1, self.stats['total_emits'])
        return {
            'total_emits': self.stats['total_emits'],
            'avg_wait_time_ms': (self.stats['lock_wait_time'] / total_emits) * 1000,
            'max_wait_time_ms': self.stats['max_wait_time'] * 1000,
            'total_wait_time_s': self.stats['lock_wait_time']
        }

class BatchingHandler(logging.Handler):
    """批量處理日志的處理器"""
    
    def __init__(self, target_handler: logging.Handler, 
                 batch_size: int = 100, 
                 flush_interval: float = 1.0):
        super().__init__()
        self.target_handler = target_handler
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        
        self.buffer = []
        self.buffer_lock = threading.Lock()
        self.last_flush = time.time()
        
        # 啟動(dòng)后臺(tái)刷新線程
        self.flush_thread = threading.Thread(target=self._flush_worker, daemon=True)
        self.flush_thread.start()
        self.shutdown_event = threading.Event()
    
    def emit(self, record):
        """批量emit實(shí)現(xiàn)"""
        with self.buffer_lock:
            self.buffer.append(record)
            
            # 檢查是否需要立即刷新
            if (len(self.buffer) >= self.batch_size or 
                time.time() - self.last_flush >= self.flush_interval):
                self._flush_buffer()
    
    def _flush_buffer(self):
        """刷新緩沖區(qū)"""
        if not self.buffer:
            return
        
        # 復(fù)制緩沖區(qū)并清空
        records_to_flush = self.buffer.copy()
        self.buffer.clear()
        self.last_flush = time.time()
        
        # 批量處理記錄
        for record in records_to_flush:
            try:
                self.target_handler.emit(record)
            except Exception:
                self.handleError(record)
    
    def _flush_worker(self):
        """后臺(tái)刷新工作線程"""
        while not self.shutdown_event.is_set():
            time.sleep(self.flush_interval)
            with self.buffer_lock:
                if self.buffer and time.time() - self.last_flush >= self.flush_interval:
                    self._flush_buffer()
    
    def close(self):
        """關(guān)閉處理器"""
        self.shutdown_event.set()
        with self.buffer_lock:
            self._flush_buffer()
        super().close()

@contextlib.contextmanager
def performance_monitor(name: str):
    """性能監(jiān)控上下文管理器"""
    start_time = time.time()
    start_memory = threading.active_count()
    
    print(f"開(kāi)始監(jiān)控: {name}")
    
    try:
        yield
    finally:
        end_time = time.time()
        end_memory = threading.active_count()
        
        print(f"監(jiān)控結(jié)束: {name}")
        print(f"執(zhí)行時(shí)間: {end_time - start_time:.3f}秒")
        print(f"線程數(shù)變化: {start_memory} -> {end_memory}")

def test_synchronization_solutions():
    """測(cè)試各種同步解決方案"""
    
    # 測(cè)試同步處理器
    base_handler = logging.FileHandler('sync_test.log')
    sync_handler = SynchronizedHandler(base_handler)
    
    logger = logging.getLogger('sync_test')
    logger.addHandler(sync_handler)
    logger.setLevel(logging.INFO)
    
    def sync_worker(worker_id):
        for i in range(50):
            logger.info(f"Sync worker {worker_id} message {i}")
            time.sleep(0.001)
    
    with performance_monitor("同步處理器測(cè)試"):
        threads = []
        for i in range(10):
            t = threading.Thread(target=sync_worker, args=(i,))
            threads.append(t)
            t.start()
        
        for t in threads:
            t.join()
    
    # 輸出性能統(tǒng)計(jì)
    stats = sync_handler.get_performance_stats()
    print(f"同步處理器統(tǒng)計(jì): {stats}")

if __name__ == "__main__":
    test_synchronization_solutions()

import asyncio
import logging
import threading
import time
from typing import Optional, Callable, Any
from concurrent.futures import ThreadPoolExecutor
import json

class AsyncLogProcessor:
    """異步日志處理器"""
    
    def __init__(self, batch_size: int = 50, flush_interval: float = 0.5):
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.log_queue = asyncio.Queue()
        self.handlers = []
        self.running = False
        self.stats = {
            'processed': 0,
            'batches': 0,
            'errors': 0
        }
    
    def add_handler(self, handler: logging.Handler):
        """添加處理器"""
        self.handlers.append(handler)
    
    async def start(self):
        """啟動(dòng)異步處理"""
        self.running = True
        await asyncio.gather(
            self._batch_processor(),
            self._periodic_flush()
        )
    
    async def stop(self):
        """停止異步處理"""
        self.running = False
        # 處理剩余的日志
        await self._flush_remaining()
    
    async def log_async(self, record: logging.LogRecord):
        """異步記錄日志"""
        await self.log_queue.put(record)
    
    async def _batch_processor(self):
        """批量處理器"""
        batch = []
        
        while self.running:
            try:
                # 收集批量記錄
                while len(batch) < self.batch_size and self.running:
                    try:
                        record = await asyncio.wait_for(
                            self.log_queue.get(), 
                            timeout=0.1
                        )
                        batch.append(record)
                    except asyncio.TimeoutError:
                        break
                
                if batch:
                    await self._process_batch(batch)
                    batch.clear()
                    
            except Exception as e:
                self.stats['errors'] += 1
                print(f"批量處理錯(cuò)誤: {e}")
    
    async def _process_batch(self, batch):
        """處理一批日志記錄"""
        self.stats['batches'] += 1
        self.stats['processed'] += len(batch)
        
        # 在線程池中處理I/O密集的日志寫(xiě)入
        loop = asyncio.get_event_loop()
        with ThreadPoolExecutor(max_workers=2) as executor:
            tasks = []
            for handler in self.handlers:
                task = loop.run_in_executor(
                    executor, 
                    self._write_batch_to_handler, 
                    handler, 
                    batch
                )
                tasks.append(task)
            
            await asyncio.gather(*tasks, return_exceptions=True)
    
    def _write_batch_to_handler(self, handler: logging.Handler, batch):
        """將批量記錄寫(xiě)入處理器"""
        for record in batch:
            try:
                handler.emit(record)
            except Exception as e:
                handler.handleError(record)
    
    async def _periodic_flush(self):
        """定期刷新"""
        while self.running:
            await asyncio.sleep(self.flush_interval)
            for handler in self.handlers:
                if hasattr(handler, 'flush'):
                    handler.flush()
    
    async def _flush_remaining(self):
        """刷新剩余日志"""
        remaining = []
        while not self.log_queue.empty():
            try:
                record = self.log_queue.get_nowait()
                remaining.append(record)
            except asyncio.QueueEmpty:
                break
        
        if remaining:
            await self._process_batch(remaining)

class AsyncLogHandler(logging.Handler):
    """異步日志處理器適配器"""
    
    def __init__(self, async_processor: AsyncLogProcessor):
        super().__init__()
        self.async_processor = async_processor
        self.loop = None
        self._setup_event_loop()
    
    def _setup_event_loop(self):
        """設(shè)置事件循環(huán)"""
        def run_async_processor():
            self.loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self.loop)
            self.loop.run_until_complete(self.async_processor.start())
        
        self.async_thread = threading.Thread(target=run_async_processor, daemon=True)
        self.async_thread.start()
        
        # 等待事件循環(huán)啟動(dòng)
        time.sleep(0.1)
    
    def emit(self, record):
        """發(fā)送日志記錄到異步處理器"""
        if self.loop and not self.loop.is_closed():
            future = asyncio.run_coroutine_threadsafe(
                self.async_processor.log_async(record), 
                self.loop
            )
            try:
                future.result(timeout=0.1)
            except Exception as e:
                self.handleError(record)
    
    def close(self):
        """關(guān)閉處理器"""
        if self.loop and not self.loop.is_closed():
            asyncio.run_coroutine_threadsafe(
                self.async_processor.stop(), 
                self.loop
            )
        super().close()

import logging
import logging.config
import os
from pathlib import Path

def create_production_logging_config():
    """創(chuàng)建生產(chǎn)環(huán)境日志配置"""
    
    log_dir = Path("logs")
    log_dir.mkdir(exist_ok=True)
    
    config = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'detailed': {
                'format': '%(asctime)s [%(process)d:%(thread)d] %(name)s %(levelname)s: %(message)s',
                'datefmt': '%Y-%m-%d %H:%M:%S'
            },
            'simple': {
                'format': '%(levelname)s: %(message)s'
            },
            'json': {
                'format': '{"timestamp": "%(asctime)s", "level": "%(levelname)s", "logger": "%(name)s", "message": "%(message)s", "thread": "%(thread)d"}',
                'datefmt': '%Y-%m-%dT%H:%M:%S'
            }
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO',
                'formatter': 'simple',
                'stream': 'ext://sys.stdout'
            },
            'file_info': {
                'class': 'logging.handlers.RotatingFileHandler',
                'level': 'INFO',
                'formatter': 'detailed',
                'filename': str(log_dir / 'app.log'),
                'maxBytes': 10485760,  # 10MB
                'backupCount': 5,
                'encoding': 'utf8'
            },
            'file_error': {
                'class': 'logging.handlers.RotatingFileHandler',
                'level': 'ERROR',
                'formatter': 'detailed',
                'filename': str(log_dir / 'error.log'),
                'maxBytes': 10485760,
                'backupCount': 10,
                'encoding': 'utf8'
            },
            'queue_handler': {
                'class': 'logging.handlers.QueueHandler',
                'queue': {
                    '()': 'queue.Queue',
                    'maxsize': 1000
                }
            }
        },
        'loggers': {
            '': {  # root logger
                'level': 'INFO',
                'handlers': ['queue_handler']
            },
            'app': {
                'level': 'DEBUG',
                'handlers': ['console', 'file_info', 'file_error'],
                'propagate': False
            },
            'performance': {
                'level': 'INFO',
                'handlers': ['file_info'],
                'propagate': False
            }
        }
    }
    
    return config

class ProductionLoggingManager:
    """生產(chǎn)環(huán)境日志管理器"""
    
    def __init__(self):
        self.config = create_production_logging_config()
        self.setup_logging()
        self.setup_queue_listener()
    
    def setup_logging(self):
        """設(shè)置日志配置"""
        logging.config.dictConfig(self.config)
    
    def setup_queue_listener(self):
        """設(shè)置隊(duì)列監(jiān)聽(tīng)器"""
        import queue
        import logging.handlers
        
        # 獲取隊(duì)列處理器
        root_logger = logging.getLogger()
        queue_handler = None
        
        for handler in root_logger.handlers:
            if isinstance(handler, logging.handlers.QueueHandler):
                queue_handler = handler
                break
        
        if queue_handler:
            # 創(chuàng)建實(shí)際的處理器
            file_handler = logging.handlers.RotatingFileHandler(
                'logs/queue_app.log',
                maxBytes=10485760,
                backupCount=5
            )
            file_handler.setFormatter(
                logging.Formatter(
                    '%(asctime)s [%(process)d:%(thread)d] %(name)s %(levelname)s: %(message)s'
                )
            )
            
            # 啟動(dòng)隊(duì)列監(jiān)聽(tīng)器
            self.queue_listener = logging.handlers.QueueListener(
                queue_handler.queue,
                file_handler,
                respect_handler_level=True
            )
            self.queue_listener.start()
    
    def get_logger(self, name: str) -> logging.Logger:
        """獲取日志器"""
        return logging.getLogger(name)
    
    def shutdown(self):
        """關(guān)閉日志系統(tǒng)"""
        if hasattr(self, 'queue_listener'):
            self.queue_listener.stop()
        logging.shutdown()

# 使用示例
def demonstrate_production_logging():
    """演示生產(chǎn)環(huán)境日志使用"""
    
    log_manager = ProductionLoggingManager()
    
    # 獲取不同類型的日志器
    app_logger = log_manager.get_logger('app.service')
    perf_logger = log_manager.get_logger('performance')
    
    def simulate_application_work():
        """模擬應(yīng)用程序工作"""
        app_logger.info("應(yīng)用程序啟動(dòng)")
        
        for i in range(100):
            app_logger.debug(f"處理任務(wù) {i}")
            
            if i % 20 == 0:
                perf_logger.info(f"性能檢查點(diǎn): 已處理 {i} 個(gè)任務(wù)")
            
            if i == 50:
                app_logger.warning("達(dá)到中間檢查點(diǎn)")
            
            # 模擬錯(cuò)誤
            if i == 75:
                try:
                    raise ValueError("模擬業(yè)務(wù)錯(cuò)誤")
                except ValueError as e:
                    app_logger.error(f"業(yè)務(wù)錯(cuò)誤: {e}", exc_info=True)
        
        app_logger.info("應(yīng)用程序完成")
    
    # 多線程測(cè)試
    threads = []
    for i in range(5):
        t = threading.Thread(target=simulate_application_work)
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    # 關(guān)閉日志系統(tǒng)
    log_manager.shutdown()

if __name__ == "__main__":
    demonstrate_production_logging()

import logging
import threading
import time
import psutil
import json
from typing import Dict, List, Any
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta

@dataclass
class LoggingMetrics:
    """日志系統(tǒng)指標(biāo)"""
    timestamp: str
    queue_size: int
    queue_capacity: int
    logs_per_second: float
    error_rate: float
    memory_usage_mb: float
    thread_count: int
    handler_stats: Dict[str, Any]

class LoggingDiagnostics:
    """日志系統(tǒng)診斷工具"""
    
    def __init__(self, monitoring_interval: float = 1.0):
        self.monitoring_interval = monitoring_interval
        self.metrics_history: List[LoggingMetrics] = []
        self.is_monitoring = False
        self.log_counter = 0
        self.error_counter = 0
        self.last_reset_time = time.time()
        
        # 監(jiān)控線程
        self.monitor_thread = None
    
    def start_monitoring(self):
        """開(kāi)始監(jiān)控"""
        self.is_monitoring = True
        self.monitor_thread = threading.Thread(target=self._monitoring_loop, daemon=True)
        self.monitor_thread.start()
        print("日志系統(tǒng)監(jiān)控已啟動(dòng)")
    
    def stop_monitoring(self):
        """停止監(jiān)控"""
        self.is_monitoring = False
        if self.monitor_thread:
            self.monitor_thread.join()
        print("日志系統(tǒng)監(jiān)控已停止")
    
    def _monitoring_loop(self):
        """監(jiān)控循環(huán)"""
        while self.is_monitoring:
            try:
                metrics = self._collect_metrics()
                self.metrics_history.append(metrics)
                
                # 保持歷史記錄在合理范圍內(nèi)
                if len(self.metrics_history) > 1000:
                    self.metrics_history = self.metrics_history[-500:]
                
                # 檢查告警條件
                self._check_alerts(metrics)
                
            except Exception as e:
                print(f"監(jiān)控錯(cuò)誤: {e}")
            
            time.sleep(self.monitoring_interval)
    
    def _collect_metrics(self) -> LoggingMetrics:
        """收集指標(biāo)"""
        current_time = time.time()
        time_diff = current_time - self.last_reset_time
        
        # 計(jì)算速率
        logs_per_second = self.log_counter / max(time_diff, 1)
        error_rate = self.error_counter / max(self.log_counter, 1)
        
        # 獲取系統(tǒng)指標(biāo)
        process = psutil.Process()
        memory_usage = process.memory_info().rss / 1024 / 1024  # MB
        thread_count = threading.active_count()
        
        # 獲取隊(duì)列信息（如果存在）
        queue_size, queue_capacity = self._get_queue_info()
        
        # 獲取處理器統(tǒng)計(jì)
        handler_stats = self._get_handler_stats()
        
        metrics = LoggingMetrics(
            timestamp=datetime.now().isoformat(),
            queue_size=queue_size,
            queue_capacity=queue_capacity,
            logs_per_second=logs_per_second,
            error_rate=error_rate,
            memory_usage_mb=memory_usage,
            thread_count=thread_count,
            handler_stats=handler_stats
        )
        
        # 重置計(jì)數(shù)器
        self.log_counter = 0
        self.error_counter = 0
        self.last_reset_time = current_time
        
        return metrics
    
    def _get_queue_info(self) -> tuple:
        """獲取隊(duì)列信息"""
        # 這里需要根據(jù)實(shí)際使用的隊(duì)列處理器來(lái)實(shí)現(xiàn)
        # 示例實(shí)現(xiàn)
        try:
            root_logger = logging.getLogger()
            for handler in root_logger.handlers:
                if hasattr(handler, 'queue'):
                    queue = handler.queue
                    if hasattr(queue, 'qsize') and hasattr(queue, 'maxsize'):
                        return queue.qsize(), queue.maxsize
            return 0, 0
        except:
            return 0, 0
    
    def _get_handler_stats(self) -> Dict[str, Any]:
        """獲取處理器統(tǒng)計(jì)信息"""
        stats = {}
        root_logger = logging.getLogger()
        
        for i, handler in enumerate(root_logger.handlers):
            handler_name = f"{type(handler).__name__}_{i}"
            handler_stats = {
                'type': type(handler).__name__,
                'level': handler.level,
                'formatter': type(handler.formatter).__name__ if handler.formatter else None
            }
            
            # 如果處理器有自定義統(tǒng)計(jì)方法
            if hasattr(handler, 'get_stats'):
                handler_stats.update(handler.get_stats())
            
            stats[handler_name] = handler_stats
        
        return stats
    
    def _check_alerts(self, metrics: LoggingMetrics):
        """檢查告警條件"""
        alerts = []
        
        # 隊(duì)列使用率告警
        if metrics.queue_capacity > 0:
            queue_usage = metrics.queue_size / metrics.queue_capacity
            if queue_usage > 0.8:
                alerts.append(f"隊(duì)列使用率過(guò)高: {queue_usage:.1%}")
        
        # 錯(cuò)誤率告警
        if metrics.error_rate > 0.05:  # 5%
            alerts.append(f"錯(cuò)誤率過(guò)高: {metrics.error_rate:.1%}")
        
        # 內(nèi)存使用告警
        if metrics.memory_usage_mb > 500:  # 500MB
            alerts.append(f"內(nèi)存使用過(guò)高: {metrics.memory_usage_mb:.1f}MB")
        
        # 線程數(shù)告警
        if metrics.thread_count > 50:
            alerts.append(f"線程數(shù)過(guò)多: {metrics.thread_count}")
        
        if alerts:
            print(f"[ALERT] {datetime.now()}: {'; '.join(alerts)}")
    
    def increment_log_count(self):
        """增加日志計(jì)數(shù)"""
        self.log_counter += 1
    
    def increment_error_count(self):
        """增加錯(cuò)誤計(jì)數(shù)"""
        self.error_counter += 1
    
    def get_recent_metrics(self, minutes: int = 5) -> List[LoggingMetrics]:
        """獲取最近的指標(biāo)"""
        cutoff_time = datetime.now() - timedelta(minutes=minutes)
        
        recent_metrics = []
        for metric in reversed(self.metrics_history):
            metric_time = datetime.fromisoformat(metric.timestamp)
            if metric_time >= cutoff_time:
                recent_metrics.append(metric)
            else:
                break
        
        return list(reversed(recent_metrics))
    
    def generate_report(self) -> str:
        """生成診斷報(bào)告"""
        if not self.metrics_history:
            return "暫無(wú)監(jiān)控?cái)?shù)據(jù)"
        
        recent_metrics = self.get_recent_metrics(10)  # 最近10分鐘
        
        if not recent_metrics:
            return "最近10分鐘無(wú)監(jiān)控?cái)?shù)據(jù)"
        
        # 計(jì)算統(tǒng)計(jì)信息
        avg_logs_per_sec = sum(m.logs_per_second for m in recent_metrics) / len(recent_metrics)
        avg_error_rate = sum(m.error_rate for m in recent_metrics) / len(recent_metrics)
        avg_memory = sum(m.memory_usage_mb for m in recent_metrics) / len(recent_metrics)
        max_queue_size = max(m.queue_size for m in recent_metrics)
        
        report = f"""
=== 日志系統(tǒng)診斷報(bào)告 ===
時(shí)間范圍: 最近10分鐘
數(shù)據(jù)點(diǎn)數(shù): {len(recent_metrics)}

性能指標(biāo):
- 平均日志速率: {avg_logs_per_sec:.2f} logs/sec
- 平均錯(cuò)誤率: {avg_error_rate:.2%}
- 平均內(nèi)存使用: {avg_memory:.1f} MB
- 最大隊(duì)列長(zhǎng)度: {max_queue_size}

當(dāng)前狀態(tài):
- 線程數(shù): {recent_metrics[-1].thread_count}
- 隊(duì)列使用: {recent_metrics[-1].queue_size}/{recent_metrics[-1].queue_capacity}
- 內(nèi)存使用: {recent_metrics[-1].memory_usage_mb:.1f} MB

處理器狀態(tài):
{json.dumps(recent_metrics[-1].handler_stats, indent=2, ensure_ascii=False)}
"""
        
        return report

class DiagnosticHandler(logging.Handler):
    """帶診斷功能的處理器包裝器"""
    
    def __init__(self, target_handler: logging.Handler, diagnostics: LoggingDiagnostics):
        super().__init__()
        self.target_handler = target_handler
        self.diagnostics = diagnostics
    
    def emit(self, record):
        """發(fā)送日志記錄"""
        try:
            self.target_handler.emit(record)
            self.diagnostics.increment_log_count()
        except Exception as e:
            self.diagnostics.increment_error_count()
            self.handleError(record)

# 使用示例
def demonstrate_logging_diagnostics():
    """演示日志診斷功能"""
    
    # 創(chuàng)建診斷工具
    diagnostics = LoggingDiagnostics(monitoring_interval=0.5)
    
    # 設(shè)置日志
    logger = logging.getLogger('diagnostic_test')
    base_handler = logging.StreamHandler()
    diagnostic_handler = DiagnosticHandler(base_handler, diagnostics)
    logger.addHandler(diagnostic_handler)
    logger.setLevel(logging.INFO)
    
    # 啟動(dòng)監(jiān)控
    diagnostics.start_monitoring()
    
    try:
        # 模擬日志活動(dòng)
        def log_worker(worker_id):
            for i in range(100):
                logger.info(f"Worker {worker_id} message {i}")
                time.sleep(0.01)
                
                # 模擬一些錯(cuò)誤
                if i % 30 == 0:
                    try:
                        raise ValueError("測(cè)試錯(cuò)誤")
                    except ValueError:
                        logger.error("模擬錯(cuò)誤", exc_info=True)
        
        # 啟動(dòng)多個(gè)工作線程
        threads = []
        for i in range(3):
            t = threading.Thread(target=log_worker, args=(i,))
            threads.append(t)
            t.start()
        
        # 等待一段時(shí)間后生成報(bào)告
        time.sleep(5)
        print(diagnostics.generate_report())
        
        # 等待所有線程完成
        for t in threads:
            t.join()
        
        # 最終報(bào)告
        print("\n=== 最終報(bào)告 ===")
        print(diagnostics.generate_report())
        
    finally:
        diagnostics.stop_monitoring()

if __name__ == "__main__":
    demonstrate_logging_diagnostics()