Socket/client_socket.py - master - dilanms/Services

Socket/client_socket.py
84 строки3 KB
Dilanms
добавление микропрограмм сокетов
05 фев 2025, 23:36
05 фев 2025, 23:3642bd46d
import asyncio
import time
import itertools
from collections import deque

class AsyncHTTPClient:
    def __init__(self, host: str, port: int, max_concurrent: int = 1000):
        self.host = host
        self.port = port
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.tasks = set()

    async def send_request(self, path: str):
        """Отправляет HTTP-запрос с контролем соединений"""
        async with self.semaphore:
            try:
                reader, writer = await asyncio.open_connection(self.host, self.port)
                request = (
                    f"GET {path} HTTP/1.1\r\n"
                    f"Host: {self.host}\r\n"
                    "Connection: close\r\n\r\n"
                )
                
                writer.write(request.encode())
                await writer.drain()
                
                response = await reader.read(4096)
                print(f"Запрос к {path} - Ответ: {response[:200]}...")
                
                writer.close()
                await writer.wait_closed()
                return True
            except Exception as e:
                print(f"Ошибка при запросе {path}: {str(e)}")
                return False
            finally:
                self.tasks.discard(asyncio.current_task())

    async def run(self, paths: list, rps: int, total_requests: int = None):
        """Запускает клиент с точным контролем RPS и параллелизма"""
        interval = 1 / rps
        request_count = 0
        start_time = time.monotonic()
        path_cycle = itertools.cycle(paths)
        
        try:
            while True:
                if total_requests and request_count >= total_requests:
                    break
                
                task = asyncio.create_task(self.send_request(next(path_cycle)))
                self.tasks.add(task)
                task.add_done_callback(self.tasks.discard)
                request_count += 1
                
                next_time = start_time + (request_count * interval)
                delay = next_time - time.monotonic()
                if delay > 0:
                    await asyncio.sleep(delay)
                
        except asyncio.CancelledError:
            pass
        finally:
            # Дожидаемся завершения всех задач
            await asyncio.gather(*self.tasks, return_exceptions=True)

async def main():
    config = {
        "host": "127.0.0.1",
        "port": 9050,
        "paths": ["/", "/test", "/api"],
        "rps": 100,  # Попробуйте 100+ RPS
        "total_requests": 500
    }
    
    client = AsyncHTTPClient(config["host"], config["port"], max_concurrent=500)
    await client.run(
        paths=config["paths"],
        rps=config["rps"],
        total_requests=config["total_requests"]
    )

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