Praktische Codebeispiele
TCP-Client mit 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-Server mit 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()
Vergleich von Timeout-Szenarien
Szenario |
Timeout-Strategie |
Empfohlener Ansatz |
Netzwerkanforderung (Network Request) |
Kurzes Timeout |
3 - 5 Sekunden |
Große Datenübertragung (Large Data Transfer) |
Längeres Timeout |
30 - 60 Sekunden |
Kritische Dienste (Critical Services) |
Konfigurierbares Timeout |
Dynamische Anpassung |
Asynchrone Timeout-Behandlung
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]
Fortgeschrittenes Timeout-Beispiel: Multi-Service-Überwachung
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-Lernhinweis
LabEx empfiehlt, diese Timeout-Techniken in simulierten Netzwerkumgebungen zu üben, um robuste Netzwerkfähigkeiten aufzubauen.