Exemples de code pratiques
Client TCP avec délai d'attente (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()
Serveur UDP avec délai d'attente (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()
Comparaison des scénarios de délai d'attente (timeout)
Scénario |
Stratégie de délai d'attente (timeout) |
Approche recommandée |
Requête réseau |
Délai d'attente court |
3 - 5 secondes |
Transfert de gros volumes de données |
Délai d'attente plus long |
30 - 60 secondes |
Services critiques |
Délai d'attente configurable |
Ajustement dynamique |
Gestion asynchrone des délais d'attente (timeouts)
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]
Exemple avancé de délai d'attente (timeout) : Surveillance multi-services
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
Astuce d'apprentissage LabEx
LabEx recommande de pratiquer ces techniques de délai d'attente (timeout) dans des environnements réseau simulés pour développer des compétences solides en réseau.