3

Есть сервер, он принимает и отправляет сообщения, с отправкой я разобрался, через цикл каждому, а вот принимать по циклу не получится, из-за того что принимать будет того кто в данный момент в цикле, остальные будут в очереди, а это не то что нужно. Какой сейчас код приема сообщения:

def Reciver():
    global conn
    while 1:
        for i in set(conn):
            try:
                data = i.recv(1024)
                if data:
                    print(data.decode())
            except:
                pass

conn - множество с подключениями. Нужно каждое сообщение которое пришло отображать на экране.

17
  • попробовал бы колбэк апи от асинкио. удобнее, но сложнее
    – eri
    Commented 27 июн. 2017 в 16:28
  • Не, ты что, я в свои 15 с трудом понял socket :) Commented 27 июн. 2017 в 16:32
  • docs.python.org/3/library/asyncio-protocol.html там снизу есть примерчики. намного проще сырого сокета
    – eri
    Commented 27 июн. 2017 в 16:36
  • Может я кину тебе свой код на socket а ты переделаешь его под asyncio? Я код-то писал дня 4, переводить все туда займет не знаю сколько еще, заодно решишь проблему с приемом сообщений и напишешь комментарии, а я почитаю, разберу некоторые функции по отдельности... Commented 27 июн. 2017 в 16:40
  • 1
    Если хотите понять как можно одновременно несколько блокирующих операций делать, посмотрите на примеры кода из презентации David Beazley - Python Concurrency From the Ground Up: LIVE! - PyCon 2015. Это для понимания. На практике, используйте библиотеки более высокого уровня вместо sockets (для разных задач много библиотек есть: requests, flask (gunicorn, uwsgi), django, gevent, tornado, twisted, aiohttp, etc)
    – jfs
    Commented 30 июн. 2017 в 17:20

1 ответ 1

9

Проблему того, что операции на сокетах блокируют поток выполнения можно решить тремя способами:

  1. Выделить для каждой блокирующейся операции отдельный поток выполнения, который она может заблокировать без вреда для остальных потоков;
  2. Перевести сокеты в неблокирующийся режим и опрашивать их состояние в цикле;
  3. Работать с сокетами в асинхронном режиме.

Ниже рассмотрены все три варианта.

Многопоточный сервер

Пожалуй, самый часто используемый способ. Можно сказать, классический. Для взаимодействия между потоками будем использовать queue.Queue. Во-первых, логика работы очереди нам как раз подходит. Во-вторых, очередь потокобезопасна.

# -*- coding: utf-8 -*-
import socket
import threading
import queue

# Определяем константу содержащую имя ОС
# для учёта особенностей данной операционной системы
import platform
OS_NAME = platform.system()

# Константы
HOST = 'localhost'
PORT = 1080

# Единственная глобальная переменная
# доступная всем потокам
run = True

def shutdown_socket(s):
    # В Linux'ах просто закрыть заблокированный сокет будет мало,
    # он так и не выйдет из состояния блокировки. Нужно передать
    # ему команду на завершение. Но в Windows наоборот, команда
    # на завершение вызовет зависание, если сокет был заблокирован
    # вызовом accept(), а простое закрытие сработает.
    if OS_NAME == 'Linux':
        s.shutdown(socket.SHUT_RDWR)
    s.close()    

def reciver(client, q):
    while run:
        try:
            # Здесь поток блокируется до тех пор
            # пока не будут считаны все имеющиеся
            # в сокете данные
            data = client.recv(1024)
            if data: # Если есть данные
                # Отправляем в очередь сообщений кортеж
                # содержащий сокет отправителя
                # и принятые данные
                q.put((client, data))
                print('{} отправил: {}'.format(client.getpeername(), data.decode()))
        except:
            break # В случае ошибки выходим из цикла
    client.close() # И закрываем клиентский сокет


def sender(q, connections):
    while run:
        closed_sockets = set()
        try:
            # Получаем из очереди сообщений
            # сокет отправителя и принятые данные
            sender, message = q.get(timeout=1)
        except queue.Empty:
            pass # Игнорируем отсутствие сообщений в очереди
        else: # Если же сообщения есть
            for s in set(connections): # Обходим все сокеты
                if s != sender: # Кроме сокета отправителя
                    try:
                        s.send(message) # И отправляем им принятое сообщение
                    except:
                        closed_sockets.add(s)
            if closed_sockets:
                with threading.Lock():
                    connections -= closed_sockets
                print("Подключено:", len(connections))
            q.task_done() # Сообщаем, что сообщение обработано


def accepter(server, connections, q):
    while run:
        try:
            # Здесь поток блокируется до тех пор, пока кто-то не подключится к серверу
            client, addr = server.accept()
        except OSError as e:
            # Если отловлена не ожидаемая ошибка закрытия серверного сокета, а какая-то другая
            if (OS_NAME == 'Windows' and e.errno != 10038) or (OS_NAME == 'Linux' and e.errno != 22):
                raise # то возбуждаем её повторно
        else: # Если кто-то подключился и создан новый клиентский сокет
            # Устанавливаем ему таймаут, чтобы считать его сбойным, 
            # если в этот сокет не будут ничего писать более 5 минут
            client.settimeout(60 * 5)
            with threading.Lock():
                connections.add(client)
            # Запускаем новый поток, выполняющий функцию receiver
            # для только что полученного сокета
            threading.Thread(target=reciver, args=(client, q)).start()
            print("Подключено:", len(connections))


if __name__ == '__main__':
    print('Запуск...')

    # Очередь сообщений, через которую будут общаться потоки
    q = queue.Queue()
    # Множество соединений
    connections = set()

    server = socket.socket()
    server.bind((HOST, PORT))
    server.listen()

    print(u'Сервер запущен на {}\n'.format(server.getsockname()))

    # Поток получающий сообщения из очереди
    # и отправляющий их всем сокетам в множестве connections
    threading.Thread(target=sender, args=(q, connections)).start()
    # Поток принимающий новые соединения
    threading.Thread(target=accepter, args=(server, connections, q)).start()

    while True:
        command = input()
        if command == 'exit': # Если в консоли введена команда exit
            run = False # отменяем выполнение циклов во всех потоках
            break # и выходим из этого цикла
    for s in connections:
        shutdown_socket(s)
    shutdown_socket(server)

Кроме внесения в ваш код изменений необходимых для работы примера, я избавился от лишних глобальных переменных и внёс стилистические правки.

Неблокирующийся сервер

Очень старый, но не потерявший своей актуальности способ. Применялся ещё древними при работе в однозадачных операционных системах. Для опроса состояния сокетов будем использовать функцию select из одноимённого модуля. Она не самая быстрая из подобных, но работает во всех операционных системах.

# -*- coding: utf-8 -*-
import select
import socket
import queue

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setblocking(0) # Неблокирующийся сокет
server.bind(('localhost', 1080))
server.listen()

sockets = [server]
message_queues = {}

def close_connection(con):
    sockets.remove(con)
    if con in message_queues:
        del message_queues[con]
    con.close()


# Пока есть хоть один сокет
while sockets:
    # Опрашиваем сокеты на готовность к чтению, записи, ошибки.
    # С таймаутом в 1 секунду для того, чтобы программа реагировала
    # на другие события.
    readable, writable, exceptional = select.select(sockets, sockets, sockets, 1)

    for s in readable: # Для каждого сокета готового к чтению
        if s is server: # Если это сокет принимающий соединения
            connection, client_address = s.accept()
            connection.setblocking(0) # Этот клиентский сокет тоже будет неблокируемым
            sockets.append(connection) # Добавляем клиентский сокет в список сокетов
            message_queues[connection] = queue.Queue() # Создаём очередь сообщений для сокета
        else:
            try:
                data = s.recv(1024) # Читаем без блокировки
            except:
                close_connection(s) # В случае ошибки закрываем этот сокет и удаляем
            else: # Если ошибка не произошла
                if data: # И данные получены
                    for c in message_queues: # Обходим все очереди сообщений
                        if c != s: # Кроме очереди текущего сокета
                            message_queues[c].put(data) # Отправляем данные в каждую очередь
                else:
                    # Если данных нет в сокете готовом для чтения
                    # значит он в состоянии закрытия на клиентской
                    # стороне. Закрываем его на стороне сервера.
                    close_connection(s)

    for s in writable: # Для каждого сокета готового к записи
        try:
            next_msg = message_queues[s].get_nowait() # Получаем сообщение из очереди
        except queue.Empty:
            pass # Игнорируем пустые очереди
        except KeyError:
            pass # Игнорируем очереди удалённые до того, как до них дошла очередь обработки
        else:
            s.send(next_msg) # Отправляем без блокировки

    for s in exceptional: # Для каждого сбойного сокета
        close_connection(s) # Закрываем сбойный сокет

Асинхронный сервер

Относительно новый способ. Asyncio предоставляет три уровня абстракции для работы с сокетами - асинхронные обёртки над операциями с сокетами, потоки и протоколы. Названия могут вызывать некоторую путаницу, потоки asyncio не имеют ничего общего с потоками выполнения, а протоколы - это не то же самое, что сетевые протоколы. При прочих равных выбирать надо всегда наиболее высокий уровень абстракции, поэтому реализуем протокол.

# -*- coding: utf-8 -*-
import asyncio

clients = []

class SimpleChatClientProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.peername = transport.get_extra_info("peername")
        print('Подключился: {}'.format(self.peername))
        clients.append(self)

    def data_received(self, data):
        print('{} отправил: {}'.format(self.peername, data.decode()))
        for client in clients:
            if client is not self:
                client.transport.write(data)

    def connection_lost(self, ex):
        print('Отключился: {}'.format(self.peername))
        clients.remove(self)


# Цикл событий невозможно прервать, если в нём
# не происходят события. Чтобы избежать этого
# регистрируем в цикле фунцию, которая будет 
# вызываться раз в секунду.
def wakeup():
    loop = asyncio.get_event_loop()
    loop.call_later(1, wakeup)


if __name__ == '__main__':
    print('Запуск...')

    # Получаем цикл событий
    loop = asyncio.get_event_loop()
    # Регистрируем "отлипатель"
    loop.call_later(1, wakeup)
    # Создаём асинхронную сопрограмму-протокол
    coro = loop.create_server(SimpleChatClientProtocol, host='localhost', port=1080)
    # Регистрируем её в цикле событий на выполнение
    server = loop.run_until_complete(coro)

    for socket in server.sockets:
        print('Сервер запущен на {}'.format(socket.getsockname()))
    print('Выход по Ctrl+C\n')

    try:
        loop.run_forever() # Запускаем бесконечный цикл событий
    except KeyboardInterrupt: # Программа прервана нажатием Ctrl+C
        pass
    finally:
        server.close() # Закрываем протокол
        loop.run_until_complete(server.wait_closed()) # Асинхронно ожидаем окончания закрытия
    loop.close() # Закрываем цикл событий

Код протокола настолько простой и очевидный, что я даже не нашёл подходящих комментариев.

С каждым из приведённых примеров можно работать как через telnet, так и с помощью вашего клиентского кода.

15

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.