Модуль socket в Python: от основ к сложным сетевым приложениям

import socket

# Создание TCP сокета
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Создание UDP сокета
udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
# Работа с сокетом
pass # Сокет будет автоматически закрыт

import socket

# Создаем серверный сокет
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Опция для повторного использования адреса
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# Привязываем сокет к адресу и порту
server_address = ('localhost', 8888)
server_socket.bind(server_address)

# Начинаем прослушивание, очередь до 5 соединений
server_socket.listen(5)
print(f"Сервер запущен на {server_address[0]}:{server_address[1]}")

try:
while True:
# Принимаем входящее соединение
client_socket, client_address = server_socket.accept()
print(f"Подключен клиент: {client_address[0]}:{client_address[1]}")

try:
# Получаем данные от клиента
data = client_socket.recv(1024)
if data:
print(f"Получены данные: {data.decode('utf-8')}")

# Отправляем ответ
response = f"Сервер получил: {data.decode('utf-8')}"
client_socket.sendall(response.encode('utf-8'))
else:
print("Клиент отключился")
finally:
# Закрываем соединение с клиентом
client_socket.close()
finally:
# Закрываем серверный сокет
server_socket.close()

import socket
import threading

def handle_client(client_socket, client_address):
try:
print(f"Обработка клиента {client_address}")
while True:
data = client_socket.recv(1024)
if not data:
break

print(f"Получено от {client_address}: {data.decode('utf-8')}")
client_socket.sendall(f"Эхо: {data.decode('utf-8')}".encode('utf-8'))
except Exception as e:
print(f"Ошибка при обработке клиента {client_address}: {e}")
finally:
client_socket.close()
print(f"Соединение с {client_address} закрыто")

def start_server():
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

server_address = ('localhost', 8888)
server_socket.bind(server_address)
server_socket.listen(10)
print(f"Сервер запущен на {server_address[0]}:{server_address[1]}")

try:
while True:
client_socket, client_address = server_socket.accept()
client_thread = threading.Thread(
target=handle_client,
args=(client_socket, client_address)
)
client_thread.daemon = True
client_thread.start()
print(f"Активные соединения: {threading.active_count() – 1}")
except KeyboardInterrupt:
print("Завершение работы сервера")
finally:
server_socket.close()

if __name__ == "__main__":
start_server()

import socket

def create_client():
# Создаем сокет
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Адрес сервера для подключения
server_address = ('localhost', 8888)

try:
# Подключаемся к серверу
print(f"Подключение к {server_address[0]}:{server_address[1]}")
client_socket.connect(server_address)

# Отправляем данные
message = "Привет, сервер!"
client_socket.sendall(message.encode('utf-8'))
print(f"Отправлено: {message}")

# Получаем ответ
data = client_socket.recv(1024)
print(f"Получено: {data.decode('utf-8')}")

except ConnectionRefusedError:
print("Сервер не доступен. Проверьте, запущен ли он.")
except Exception as e:
print(f"Произошла ошибка: {e}")
finally:
# Закрываем соединение
client_socket.close()
print("Соединение закрыто")

if __name__ == "__main__":
create_client()

import socket
import time

def receive_all(sock, length):
"""Функция для приема точного количества байт"""
data = b''
while len(data) < length:
more = sock.recv(length – len(data))
if not more:
raise ConnectionError("Соединение прервано")
data += more
return data

def send_message(sock, message):
"""Отправка сообщения с префиксом длины"""
encoded_message = message.encode('utf-8')
length_prefix = len(encoded_message).to_bytes(4, byteorder='big')
sock.sendall(length_prefix + encoded_message)

def receive_message(sock):
"""Получение сообщения с префиксом длины"""
# Получаем 4 байта префикса длины
length_prefix = receive_all(sock, 4)
message_length = int.from_bytes(length_prefix, byteorder='big')

# Получаем сообщение указанной длины
message_data = receive_all(sock, message_length)
return message_data.decode('utf-8')

def create_robust_client():
# Максимальное количество попыток подключения
max_attempts = 5
attempt = 0

while attempt < max_attempts:
attempt += 1

client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.settimeout(10) # Тайм-аут 10 секунд

server_address = ('localhost', 8888)

try:
print(f"Попытка подключения {attempt}/{max_attempts}")
client_socket.connect(server_address)

# Отправляем сообщение
message = "Это тестовое сообщение от клиента!"
send_message(client_socket, message)
print(f"Отправлено: {message}")

# Получаем ответ
response = receive_message(client_socket)
print(f"Получено: {response}")

# Успешное завершение, выходим из цикла
break

except ConnectionRefusedError:
print("Сервер не доступен")
if attempt < max_attempts:
print(f"Повторная попытка через 2 секунды...")
time.sleep(2)
except socket.timeout:
print("Тайм-аут операции")
if attempt < max_attempts:
print(f"Повторная попытка через 2 секунды...")
time.sleep(2)
except Exception as e:
print(f"Ошибка: {e}")
if attempt < max_attempts:
print(f"Повторная попытка через 2 секунды...")
time.sleep(2)
finally:
client_socket.close()

if attempt == max_attempts:
print("Все попытки подключения исчерпаны")

if __name__ == "__main__":
create_robust_client()

import socket

def start_udp_server():
# Создаем UDP сокет
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# Привязываем к адресу и порту
server_address = ('localhost', 9999)
server_socket.bind(server_address)
print(f"UDP сервер запущен на {server_address[0]}:{server_address[1]}")

try:
while True:
# Получаем данные и адрес клиента
data, client_address = server_socket.recvfrom(1024)
print(f"Получено {len(data)} байт от {client_address}")
print(f"Данные: {data.decode('utf-8')}")

# Отправляем ответ клиенту
response = f"Эхо: {data.decode('utf-8')}"
server_socket.sendto(response.encode('utf-8'), client_address)
print(f"Ответ отправлен")
except KeyboardInterrupt:
print("Сервер остановлен")
finally:
server_socket.close()

if __name__ == "__main__":
start_udp_server()

import socket

def udp_client():
# Создаем UDP сокет
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# Устанавливаем тайм-аут
client_socket.settimeout(5)

# Адрес сервера
server_address = ('localhost', 9999)

try:
# Отправляем сообщение
message = "Привет, UDP сервер!"
client_socket.sendto(message.encode('utf-8'), server_address)
print(f"Отправлено: {message}")

# Получаем ответ
data, server = client_socket.recvfrom(1024)
print(f"Получено от {server}: {data.decode('utf-8')}")

except socket.timeout:
print("Тайм-аут при ожидании ответа")
except Exception as e:
print(f"Ошибка: {e}")
finally:
client_socket.close()

if __name__ == "__main__":
udp_client()

import socket
import time
import random
import struct

class ReliableUDPClient:
def __init__(self, server_address, timeout=5.0, retries=3):
self.server_address = server_address
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.socket.settimeout(timeout)
self.retries = retries
self.sequence_number = 0

def send_reliable(self, message):
data = message.encode('utf-8')
message_id = random.randint(1, 65535)

# Формат: message_id (2 байта), sequence_number (2 байта), data
header = struct.pack('!HH', message_id, self.sequence_number)
packet = header + data

for attempt in range(self.retries):
try:
# Отправка пакета
self.socket.sendto(packet, self.server_address)
print(f"Отправлен пакет #{self.sequence_number}, попытка {attempt+1}")

# Ожидание подтверждения
while True:
response, _ = self.socket.recvfrom(1024)

# Разбор заголовка ответа
ack_id, ack_seq = struct.unpack('!HH', response[:4])

# Проверка соответствия сообщения
if ack_id == message_id and ack_seq == self.sequence_number:
print(f"Получено подтверждение для пакета #{self.sequence_number}")
self.sequence_number = (self.sequence_number + 1) % 65536
return response[4:].decode('utf-8')

except socket.timeout:
print(f"Тайм-аут при ожидании подтверждения, повтор...")
continue

raise TimeoutError(f"Не удалось получить подтверждение после {self.retries} попыток")

def close(self):
self.socket.close()

# Пример использования
if __name__ == "__main__":
client = ReliableUDPClient(('localhost', 9999))
try:
response = client.send_reliable("Тестовое сообщение через надежный UDP")
print(f"Ответ: {response}")
finally:
client.close()

import socket
import select

# Создаем неблокирующий сокет
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setblocking(False) # Устанавливаем неблокирующий режим
server_socket.bind(('localhost', 8888))
server_socket.listen(5)

# Списки для мониторинга
inputs = [server_socket] # Сокеты, которые ожидаем читать
outputs = [] # Сокеты, готовые для записи
message_queues = {} # Сообщения для отправки

while inputs:
# select() блокирует выполнение до готовности сокетов
readable, writable, exceptional = select.select(inputs, outputs, inputs)

# Обрабатываем сокеты, готовые для чтения
for s in readable:
if s is server_socket:
# Основной сервер готов принять соединение
client_socket, client_address = s.accept()
client_socket.setblocking(False)
inputs.append(client_socket)
message_queues[client_socket] = []
print(f"Новое соединение: {client_address}")
else:
# Клиентский сокет готов для чтения
try:
data = s.recv(1024)
if data:
# Клиент отправил данные
print(f"Получено от {s.getpeername()}: {data.decode('utf-8')}")
message_queues[s].append(data)
if s not in outputs:
outputs.append(s)
else:
# Клиент отключился
print(f"Клиент {s.getpeername()} отключился")
if s in outputs:
outputs.remove(s)
inputs.remove(s)
s.close()
del message_queues[s]
except ConnectionError:
# Обработка ошибок соединения
print(f"Ошибка соединения с {s.getpeername()}")
inputs.remove(s)
if s in outputs:
outputs.remove(s)
s.close()
del message_queues[s]

# Обрабатываем сокеты, готовые для записи
for s in writable:
if message_queues[s]:
# Отправляем первое сообщение из очереди
next_msg = message_queues[s].pop(0)
response = f"Эхо: {next_msg.decode('utf-8')}".encode('utf-8')
s.send(response)
else:
# Нет данных для отправки
outputs.remove(s)

# Обрабатываем исключительные ситуации
for s in exceptional:
print(f"Исключительная ситуация для {s.getpeername()}")
inputs.remove(s)
if s in outputs:
outputs.remove(s)
s.close()
del message_queues[s]

import socket

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Повторное использование адреса
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# Отключение алгоритма Нейгла для TCP (уменьшение задержки)
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

# Установка тайм-аута операций
s.settimeout(10.0) # 10 секунд

# Размер буфера приема
s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)

# Размер буфера отправки
s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)

# Включение keepalive
s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

import socket

# Получение IP по имени хоста
ip_address = socket.gethostbyname('www.example.com')
print(f"IP адрес: {ip_address}")

# Получение расширенной информации о хосте
host_info = socket.gethostbyname_ex('www.example.com')
print(f"Имя хоста: {host_info[0]}")
print(f"Список IP: {host_info[2]}")

# Получение имени хоста по IP
try:
host_name = socket.gethostbyaddr('93.184.216.34')[0]
print(f"Имя хоста: {host_name}")
except socket.herror:
print("Не удалось получить имя хоста")

# Преобразование имени и порта в кортеж адреса
address_info = socket.getaddrinfo(
'www.example.com', 80, 
socket.AF_INET, socket.SOCK_STREAM
)
for info in address_info:
family, socktype, proto, canonname, sockaddr = info
print(f"Семейство: {family}, Тип: {socktype}, Адрес: {sockaddr}")

# Получение информации о сервисе по имени
try:
port = socket.getservbyname('http')
print(f"Порт HTTP: {port}")
except OSError:
print("Сервис не найден")

import asyncio

async def handle_client(reader, writer):
# Получение адреса клиента
addr = writer.get_extra_info('peername')
print(f"Подключен клиент: {addr}")

while True:
# Асинхронное чтение данных
data = await reader.read(1024)

if not data:
break

message = data.decode('utf-8')
print(f"Получено от {addr}: {message}")

# Отправка ответа
response = f"Эхо: {message}"
writer.write(response.encode('utf-8'))
await writer.drain() # Ожидание завершения записи

print(f"Клиент {addr} отключился")
writer.close()
await writer.wait_closed() # Ожидание закрытия соединения

async def start_server():
# Создание асинхронного TCP-сервера
server = await asyncio.start_server(
handle_client, 'localhost', 8888
)

addr = server.sockets[0].getsockname()
print(f"Сервер запущен на {addr}")

async with server:
await server.serve_forever()

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

import socket
import ssl

# Создание SSL контекста
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain(certfile="server.crt", keyfile="server.key")

# Создание обычного сокета
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('localhost', 8443))
server_socket.listen(5)

while True:
# Принимаем соединение
client_socket, client_address = server_socket.accept()
print(f"Подключен клиент: {client_address}")

try:
# Оборачиваем сокет в SSL
secure_socket = context.wrap_socket(
client_socket, server_side=True
)

# Теперь работаем с защищенным сокетом
data = secure_socket.recv(1024)
print(f"Получено: {data.decode('utf-8')}")

secure_socket.send("Защищенный ответ".encode('utf-8'))
except ssl.SSLError as e:
print(f"Ошибка SSL: {e}")
finally:
client_socket.close()