Практические примеры кода
TCP-клиент с таймаутом (timeout)
import socket
def tcp_client_with_timeout(host, port, timeout=5):
try:
## Create socket with timeout
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.settimeout(timeout)
## Attempt connection
client_socket.connect((host, port))
client_socket.send(b'Hello Server')
## Receive data with timeout
response = client_socket.recv(1024)
print(f"Server response: {response.decode()}")
except socket.timeout:
print("Connection timed out")
except ConnectionRefusedError:
print("Connection refused")
finally:
client_socket.close()
UDP-сервер с таймаутом (timeout)
import socket
def udp_server_with_timeout(host='localhost', port=12345, timeout=10):
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind((host, port))
server_socket.settimeout(timeout)
try:
while True:
try:
data, client_address = server_socket.recvfrom(1024)
print(f"Received: {data.decode()} from {client_address}")
except socket.timeout:
print("No data received within timeout")
except KeyboardInterrupt:
print("Server stopped")
finally:
server_socket.close()
Сравнение сценариев таймаутов (timeout)
Сценарий |
Стратегия таймаута (timeout) |
Рекомендуемый подход |
Сетевой запрос |
Короткий таймаут (timeout) |
3 - 5 секунд |
Передача больших объемов данных |
Длинный таймаут (timeout) |
30 - 60 секунд |
Критические сервисы |
Настраиваемый таймаут (timeout) |
Динамическая настройка |
Асинхронная обработка таймаутов (timeout)
graph TD
A[Network Operation] --> B{Timeout Configured}
B -->|Yes| C[Start Async Timer]
C --> D{Operation Complete}
D -->|No| E{Timeout Reached}
E -->|Yes| F[Cancel Operation]
E -->|No| D
D -->|Yes| G[Process Result]
Продвинутый пример таймаута (timeout): Мониторинг нескольких сервисов
import socket
import concurrent.futures
def check_service(host, port, timeout=5):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
result = sock.connect_ex((host, port))
return f"{host}:{port} - {'Open' if result == 0 else 'Closed'}"
except socket.timeout:
return f"{host}:{port} - Timeout"
finally:
sock.close()
def multi_service_check(services):
with concurrent.futures.ThreadPoolExecutor() as executor:
results = list(executor.map(lambda s: check_service(*s), services))
return results
Совет по обучению от LabEx
LabEx рекомендует практиковать эти техники работы с таймаутами (timeout) в имитированных сетевых средах, чтобы развить надежные навыки сетевого программирования.