Python平行應用程式偵錯與監控

import pydevd_pycharm
pydevd_pycharm.settrace('your.remote.host', port=5678, stdoutToServer=True, stderrToServer=True, suspend=False)

import sys
import threading

def trace_calls(frame, event, arg):
    if event == 'call':
        code = frame.f_code
        func_name = code.co_name
        thread_name = threading.current_thread().name
        print(f"{thread_name} - call to {func_name}")
    return trace_calls

sys.settrace(trace_calls)

def sample_function():
    pass

sample_function()

import logging
import threading

log_format = '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s'
logging.basicConfig(level=logging.DEBUG, format=log_format)

def worker():
    logging.info("Worker started processing.")
    # 模擬一些處理
    logging.info("Worker completed processing.")

threads = [threading.Thread(target=worker, name=f'Thread-{i}') for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

import logging
import logging.handlers
import queue
import threading

log_queue = queue.Queue(-1)
queue_handler = logging.handlers.QueueHandler(log_queue)

logger = logging.getLogger()
logger.addHandler(queue_handler)
logger.setLevel(logging.DEBUG)

stream_handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(threadName)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)

queue_listener = logging.handlers.QueueListener(log_queue, stream_handler)
queue_listener.start()

def worker():
    logging.info("Asynchronously logging message from worker thread.")

threads = [threading.Thread(target=worker, name=f'AsyncThread-{i}') for i in range(5)]
for t in threads:
    t.start()

from prometheus_client import start_http_server, Summary, Counter
import time
import random

# 定義指標
REQUEST_TIME = Summary('request_processing_seconds', '處理請求所花費的時間')
REQUEST_COUNTER = Counter('request_count', '處理的請求總數')

@REQUEST_TIME.time()
def process_request():
    time.sleep(random.random())
    REQUEST_COUNTER.inc()

if __name__ == '__main__':
    start_http_server(8000)
    while True:
        process_request()

import logging
import json
import threading
import uuid

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            'time': self.formatTime(record, self.datefmt),
            'thread': record.threadName,
            'level': record.levelname,
            'message': record.getMessage(),
            'transaction_id': getattr(record, 'transaction_id', None)
        }
        return json.dumps(log_record)

handler = logging.StreamHandler()
formatter = JsonFormatter()
handler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)

def perform_task():
    transaction_id = str(uuid.uuid4())
    extra = {'transaction_id': transaction_id}
    logger.info("開始任務執行。", extra=extra)
    # 模擬任務工作
    logger.info("任務成功完成。", extra=extra)

threads = [threading.Thread(target=perform_task, name=f'JSONThread-{i}') for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

import psutil
import time

def monitor_resources():
    while True:
        cpu_usage = psutil.cpu_percent()
        memory_info = psutil.virtual_memory()
        print(f"CPU 使用率: {cpu_usage}% - 記憶體使用率: {memory_info.percent}%")
        time.sleep(5)

if __name__ == '__main__':
    monitor_resources()

import threading
import unittest

class ConcurrentCounter:
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            current = self.value
            # 為測試同步而設定的訊號點
            threading.Event().wait(0.001)
            self.value = current + 1

def worker(counter, iterations):
    for _ in range(iterations):
        counter.increment()

class TestConcurrentCounter(unittest.TestCase):
    def test_counter_incrementation(self):
        counter = ConcurrentCounter()
        iterations = 1000
        threads = [threading.Thread(target=worker, args=(counter, iterations)) for _ in range(4)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        self.assertEqual(counter.value, 4 * iterations)

if __name__ == '__main__':
    unittest.main()

import asyncio
import unittest

class AsyncCounter:
    def __init__(self):
        self.value = 0
        self._lock = asyncio.Lock()

    async def increment(self):
        async with self._lock:
            temp = self.value
            # 模擬非同步延遲
            await asyncio.sleep(0.001)
            self.value = temp + 1

async def async_worker(counter, iterations):
    for _ in range(iterations):
        await counter.increment()

class TestAsyncCounter(unittest.IsolatedAsyncioTestCase):
    async def test_async_counter_incrementation(self):
        counter = AsyncCounter()
        iterations = 1000
        tasks = [async_worker(counter, iterations) for _ in range(4)]
        await asyncio.gather(*tasks)
        self.assertEqual(counter.value, 4 * iterations)

if __name__ == '__main__':
    unittest.main()