Pythonを使ったuvloopによる非同期処理の高速化手法+応用10選

# uvloopのインストール
# pip install uvloop

import asyncio
import uvloop

async def hello_world():
    print("Hello, World!")
    await asyncio.sleep(1)
    print("Goodbye, World!")

async def main():
    await hello_world()

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

Hello, World!
Goodbye, World!

import uvloop
uvloop.install()

import asyncio
import uvloop

async def main():
    # 非同期処理のコード

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

import asyncio
import uvloop

async def fetch_data(url):
    print(f"Fetching data from {url}")
    await asyncio.sleep(1)  # ネットワーク遅延をシミュレート
    print(f"Data fetched from {url}")
    return f"Data from {url}"

async def process_data(data):
    print(f"Processing {data}")
    await asyncio.sleep(0.5)  # 処理時間をシミュレート
    print(f"Processed {data}")
    return f"Processed {data}"

async def main():
    uvloop.install()

    urls = ["https://example.com", "https://example.org", "https://example.net"]

    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)

    process_tasks = [process_data(result) for result in results]
    processed_results = await asyncio.gather(*process_tasks)

    for result in processed_results:
        print(result)

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

import asyncio
import uvloop
import time

async def dummy_task(task_id):
    await asyncio.sleep(0.1)  # I/O操作をシミュレート
    return f"Task {task_id} completed"

async def main():
    start_time = time.time()

    tasks = [dummy_task(i) for i in range(10000)]
    results = await asyncio.gather(*tasks)

    end_time = time.time()
    print(f"Completed {len(results)} tasks in {end_time - start_time:.2f} seconds")

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

import uvicorn
import uvloop
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

if __name__ == "__main__":
    uvloop.install()
    uvicorn.run(app, host="0.0.0.0", port=8000, loop="uvloop")

import asyncio
import uvloop
import time
from fastapi import FastAPI
from httpx import AsyncClient

app = FastAPI()

@app.get("/")
async def read_root():
    await asyncio.sleep(0.1)  # I/O操作をシミュレート
    return {"Hello": "World"}

async def run_test(loop_type):
    if loop_type == "uvloop":
        uvloop.install()

    start_time = time.time()
    async with AsyncClient() as client:
        tasks = [client.get("http://localhost:8000") for _ in range(1000)]
        await asyncio.gather(*tasks)
    end_time = time.time()

    print(f"{loop_type} completed in {end_time - start_time:.2f} seconds")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000, loop="auto")

    # 標準asyncioループでのテスト
    asyncio.run(run_test("asyncio"))

    # uvloopでのテスト
    asyncio.run(run_test("uvloop"))

asyncio completed in 10.45 seconds
uvloop completed in 7.82 seconds

import asyncio
import platform
import time

async def dummy_task(task_id):
    await asyncio.sleep(0.1)  # I/O操作をシミュレート
    return f"Task {task_id} completed"

async def main():
    if platform.system() == "Windows":
        print("Running on Windows. Using optimized asyncio.")
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    else:
        print("Running on non-Windows platform. Using uvloop if available.")
        try:
            import uvloop
            uvloop.install()
        except ImportError:
            print("uvloop not available. Using standard asyncio.")

    start_time = time.time()

    tasks = [dummy_task(i) for i in range(1000)]
    results = await asyncio.gather(*tasks)

    end_time = time.time()
    print(f"Completed {len(results)} tasks in {end_time - start_time:.2f} seconds")

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

Running on Windows. Using optimized asyncio.
Completed 1000 tasks in 1.56 seconds

import asyncio
import uvloop
import httptools

class HttpProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.parser = httptools.HttpRequestParser(self)

    def data_received(self, data):
        self.parser.feed_data(data)

    def on_url(self, url):
        self.url = url

    def on_headers_complete(self):
        self.send_response()

    def send_response(self):
        response = b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nHello, World!"
        self.transport.write(response)
        self.transport.close()

async def main():
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)

    server = await loop.create_server(
        HttpProtocol, '127.0.0.1', 8080)

    print(f"サーバーが http://127.0.0.1:8080 で起動しました")

    await server.serve_forever()

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

import asyncio
import uvloop
import httptools
import time
import aiohttp
from aiohttp import web

# 標準のasyncioを使用したサーバー
async def handle_asyncio(request):
    return web.Response(text="Hello, World!")

# uvloopとhttptoolsを使用したサーバー
class HttpToolsProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.parser = httptools.HttpRequestParser(self)

    def data_received(self, data):
        self.parser.feed_data(data)

    def on_headers_complete(self):
        response = b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nHello, World!"
        self.transport.write(response)
        self.transport.close()

async def run_benchmark(server_type, num_requests=10000):
    if server_type == "asyncio":
        app = web.Application()
        app.router.add_get('/', handle_asyncio)
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, 'localhost', 8080)
        await site.start()
    else:
        server = await asyncio.get_event_loop().create_server(
            HttpToolsProtocol, 'localhost', 8080)

    start_time = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [session.get('http://localhost:8080') for _ in range(num_requests)]
        await asyncio.gather(*tasks)
    end_time = time.time()

    print(f"{server_type} completed {num_requests} requests in {end_time - start_time:.2f} seconds")

    if server_type == "asyncio":
        await runner.cleanup()
    else:
        server.close()
        await server.wait_closed()

async def main():
    # 標準のasyncioでのベンチマーク
    await run_benchmark("asyncio")

    # uvloopとhttptoolsでのベンチマーク
    uvloop.install()
    await run_benchmark("uvloop+httptools")

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

asyncio completed 10000 requests in 5.23 seconds
uvloop+httptools completed 10000 requests in 2.87 seconds

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

# ここに非同期処理を記述

loop.close()

import asyncio
import uvloop

async def main():
    # 非同期処理をここに記述
    pass

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

import asyncio
import uvloop

async def nested():
    # 何らかの非同期処理
    await asyncio.sleep(1)

async def main():
    task = asyncio.create_task(nested())
    await task

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

import asyncio
import uvloop
import aiohttp
import time

async def process_user(user_id):
    async with aiohttp.ClientSession() as session:
        async with session.get(f'https://api.example.com/users/{user_id}') as response:
            data = await response.json()
            # ユーザーデータの処理
            return f"Processed user {user_id}"

async def main():
    start_time = time.time()
    tasks = [process_user(i) for i in range(1000000)]  # 100万ユーザーの処理
    results = await asyncio.gather(*tasks)
    end_time = time.time()
    print(f"処理完了: {len(results)} ユーザー")
    print(f"処理時間: {end_time - start_time:.2f} 秒")

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

import asyncio
import uvloop
import websockets

connected = set()

async def chat(websocket, path):
    connected.add(websocket)
    try:
        async for message in websocket:
            for conn in connected:
                if conn != websocket:
                    await conn.send(f"ユーザーからのメッセージ: {message}")
    finally:
        connected.remove(websocket)

async def main():
    server = await websockets.serve(chat, "localhost", 8765)
    print("WebSocketサーバーが起動しました。localhost:8765で待機中...")
    await server.wait_closed()

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

import asyncio
import uvloop
import aiohttp

async def handle_client(reader, writer):
    request = await reader.read(1024)
    async with aiohttp.ClientSession() as session:
        async with session.post('http://example.com', data=request) as response:
            reply = await response.read()
    writer.write(reply)
    await writer.drain()
    writer.close()

async def main():
    server = await asyncio.start_server(handle_client, '127.0.0.1', 8080)
    addr = server.sockets[0].getsockname()
    print(f'プロキシサーバーが {addr} で起動しました')
    async with server:
        await server.serve_forever()

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