分散式系統非同步程式設計應用

import aiohttp
import asyncio

async def fetch_data(session, url):
    async with session.get(url) as response:
        return await response.text()

async def handle_request(request):
    async with aiohttp.ClientSession() as session:
        data = await fetch_data(session, 'http://example.com/data')
        return aiohttp.web.Response(text=data)

app = aiohttp.web.Application()
app.router.add_get('/', handle_request)

async def main():
    runner = aiohttp.web.AppRunner(app)
    await runner.setup()
    site = aiohttp.web.TCPSite(runner, 'localhost', 8080)
    await site.start()

asyncio.run(main())

from aiohttp import web, ClientSession
import asyncio

async def fetch_data(session: ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return await response.json()

async def handle_request(request: web.Request) -> web.Response:
    async with ClientSession() as session:
        data = await fetch_data(session, 'http://remote-service/api/data')
        processed_data = await process_data(data)
        return web.json_response(processed_data)

async def process_data(data: dict) -> dict:
    await asyncio.sleep(0.1)  # 模擬非同步處理
    return {k: v * 2 for k, v in data.items()}

app = web.Application()
app.router.add_get('/', handle_request)

if __name__ == '__main__':
    web.run_app(app, port=8080)

import asyncio
import aio_pika

async def process_message(message: aio_pika.IncomingMessage) -> None:
    async with message.process():
        payload = message.body.decode()
        print(f"Processing message: {payload}")
        await asyncio.sleep(0.1)  # 模擬處理時間

async def consume(queue_name: str) -> None:
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    channel = await connection.channel()
    queue = await channel.declare_queue(queue_name, durable=True)
    await queue.consume(process_message)
    print("Consuming messages...")
    return connection

async def main():
    connection = await consume("distributed_queue")
    try:
        await asyncio.Future()
    finally:
        await connection.close()

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from aiohttp import ClientSession, ClientError

class AsyncCircuitBreaker:
    def __init__(self, threshold: int, recovery_timeout: float):
        self.failure_count = 0
        self.threshold = threshold
        self.recovery_timeout = recovery_timeout
        self.state = "CLOSED"
        self.last_failure_time = None

    async def call(self, coro):
        if self.state == "OPEN":
            if (asyncio.get_event_loop().time() - self.last_failure_time) > self.recovery_timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit Breaker is OPEN")
        try:
            result = await coro
            self.failure_count = 0
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = asyncio.get_event_loop().time()
            if self.failure_count >= self.threshold:
                self.state = "OPEN"
            raise e

async def fetch_with_cb(url: str, cb: AsyncCircuitBreaker) -> dict:
    async with ClientSession() as session:
        async def fetch():
            async with session.get(url) as response:
                return await response.json()
        return await cb.call(fetch())

async def main():
    cb = AsyncCircuitBreaker(threshold=3, recovery_timeout=5)
    try:
        data = await fetch_with_cb("http://remote-service/api/info", cb)
        print("Fetched data:", data)
    except Exception as ex:
        print("Request failed:", ex)

if __name__ == "__main__":
    asyncio.run(main())

import asyncio

shared_config = {"api_key": "secret_value"}
config_lock = asyncio.Lock()

async def read_config(key: str) -> str:
    async with config_lock:
        return shared_config.get(key, "")

async def update_config(key: str, value: str) -> None:
    async with config_lock:
        shared_config[key] = value

async def config_worker():
    # 安全地讀取和更新組態
    api_key = await read_config("api_key")
    print("Current API Key:", api_key)
    await update_config("api_key", "new_secret_value")

asyncio.run(config_worker())

from aiohttp import web
import asyncio
import jwt

SECRET = "very_secret_key"

@web.middleware
async def auth_middleware(request: web.Request, handler):
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    if not token:
        return web.json_response({"error": "Missing token"}, status=401)
    try:
        # 非同步解碼令牌（模擬密集的加密操作）
        decoded = await asyncio.to_thread(jwt.decode, token, SECRET, algorithms=["HS256"])
        request["user"] = decoded.get("user")
    except jwt.PyJWTError as ex:
        return web.json_response({"error": "Invalid token"}, status=403)
    return await handler(request)

async def protected_handler(request: web.Request) -> web.Response:
    user = request.get("user", "anonymous")
    return web.json_response({"message": f"Hello, {user}"})

app = web.Application(middlewares=[auth_middleware])
app.router.add_get("/protected", protected_handler)

if __name__ == "__main__":
    web.run_app(app, port=8081)