Сеансы многопроцессорной обработки Python

Question

У меня есть скрипт, получающий данные из сокета, каждый из которых содержит идентификатор сеанса, который нужно отслеживать, перед каждым приходящим сообщением, я открываю новый процесс с модулем многопроцессорности, у меня возникают проблемы, чтобы найти способ отслеживать новые входящие сообщения, имеющие тот же sessionid. Например:

100100 | Привет -

(открыть новый процесс)

100100 | Привет тебе

(ответы о процессе)

100101 | Hello

(открыть новый процесс)

100101 | Привет тебе

(новые ответы о процессе)

100100 | wasap? -

(открыть новый процесс)

когда приходит новое сообщение от sessionid 100100 ... как я могу отправить его процессу, который обрабатывает эти конкретные сообщения?

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

Мне нужно какое-то руководство, потому что раньше никогда не работало с многопроцессорной обработкой ...

Спасибо ...

Answer 1

Для взаимодействия с процессами, созданными с помощью multiprocessing, вы можете использовать классы Queue и Pipe (также из многопроцессорного модуля). Вот краткий пример использования Queue для отправки сообщения процессу:

from multiprocessing import Process, Queue

def f(q):
    print 'f(), waiting...'
    print q.get()

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    q.put('Hello from main!')
    p.join()

Более подробную информацию можно найти в Документах Python .

Answer 2

Я не уверен в необходимости открытия отдельного процесса для каждого сообщения, полученного из сокетов. Я полагаю, у вас есть причина сделать все, что вы упомянули.

Насколько я понимаю, вы написали сокет на стороне сервера, который прослушивает клиентское соединение, принимает клиентское соединение, получает данные и отправляет эти данные различным процессам, которые вы создали для обработки этих сообщений.

Я предполагаю, что идентификатор сеанса остается неизменным для клиентского соединения. Каждый клиент отправляет данные, помеченные его идентификатором сеанса. Если это так, то вы можете легко решить это благодаря тому, что в Unix - разветвляются дубликаты дескрипторов файлов / сокетов. когда вы запускаете multiprocessing.Process (), вы можете передать ему дескриптор сокета. Убедитесь, что вы закрываете сокет в родительском процессе или процессе, в котором вы прослушиваете клиентские соединения. Таким образом, каждый процесс будет обрабатывать соединение и будет получать только сообщения для одного клиента, помеченного тем же идентификатором сеанса.

Я надеюсь, что это отвечает вашим потребностям.

Answer 3

Как насчет хранения процессов в словаре? Например. в псевдокоде:

def __init__(self):
    self.sessions = {}
def handle(self, sessionid, data):
    proc = self.sessions.get(sessionid)
    if prod is None:
        proc = self.create_process()
        self.sessions[sessionid] = proc
    proc.handle(data)