异步、多处理和 websockets 不返回结果

Question

我正在尝试让 websockets、asyncio 和多​​进程一起工作。我已经坚持了 2 天，希望能得到一些帮助。

我在 stackoverflow 上搜索了 websockets asyncio 和 multiprocessing 以及一般的互联网搜索。我找到了线程示例，我可以使用它们。

import asyncio
import websockets
import threading


class Connection():
    def __init__(self):
        self.loop = asyncio.new_event_loop()
        sock_thread = threading.Thread(target=self.new_loop) 
        sock_thread.start()
        self.x = 0

    async def connect_to_socket(self):
        self.websocket = await websockets.connect('ws://demos.kaazing.com/echo')
        await self.websocket.send("hello")
        response = await self.websocket.recv()
        print(response)


    async def listen_to_socket(self):
         while True:
            await asyncio.sleep(0)
            print('Listening for a message...')
            while self.x < 5:
                message = await self.websocket.recv()
                print("< {}".format(message))
                print('\n\n')
                print(self.x)
                self.x += 1
            self.task.cancel()
            self.loop.close()


    def stop(self):
        print('canceling task\n\n')
        self.x = 0
        self.task.cancel()


    def new_loop(self):
        self.task = self.loop.create_task(self.connect_to_socket())
        self.loop.run_forever()


    def make_task(self):
        self.task = self.loop.create_task(self.listen_to_socket())


if __name__ == '__main__':
    conn=Connection()

这没有问题。我见过多处理在事件循环中打开一个进程的例子，这不是我想要的。我想开 但是，这不是我想要的。我想打开一个新进程并在新进程中运行一个事件循环。在事件循环中，我想运行我的套接字。我想将我的主进程从监听套接字中解放出来，并使用一个子进程来监听套接字，同时我在我的主进程上执行计算成本高昂的工作。

当我尝试以下代码时。我什么都得不到。

import asyncio
import websockets
import multiprocessing


class Connection(multiprocessing.Process):
    def __init__(self, tasks, results):
        super().__init__()
        self.tasks = tasks
        self.results = results
        self.loop = asyncio.new_event_loop()
        print('create event loop')
        self.x = 0
        self.task = self.loop.create_task(self.test())
        print('done with connecting')


    #connect to socket and get response
    async def test(self):
        self.ws = await websockets.connect('ws://demos.kaazing.com/echo')
        await self.websocket.send("hello")
        response = await self.websocket.recv()
        print(response)


    #listen to socket long term after connection
    async def listen_to_socket(self):

        while True:
           await asyncio.sleep(0)

           print('Listening for a message...')
           while self.x < 5:
               await self.websocket.send("hello")
               message = await self.websocket.recv()
               print("< {}".format(message))
               print('\n\n')
               print(self.x)
               self.x += 1
               self.results.put(message)
           self.task.cancel()
           self.loop.close()


    #stop task
    def stop(self):
        print('canceling task\n\n')
        self.x = 0
        self.task.cancel()


    # listen to socket long term
    #I have not called this as I can't even get a response from test()
    def make_task(self):
        self.task = self.loop.create_task(self.listen_to_socket())

if __name__ == '__main__':

    tasks = multiprocessing.JoinableQueue()
    results = multiprocessing.Queue()
    process = Connection(tasks, results)
    if tasks.empty():
        print('empty')

    else: 
        print(tasks.get())

我希望与套接字连接并收到响应。然而，我什么也得不到。我没有收到错误消息，连接没有打印输出，我得到一个空队列，仅此而已。如何从我的 websocket 获取返回值？ 我还不够新，我不确定我做错了什么。任何建议都会帮助我。 谢谢

Answer 1

任何有兴趣的人，我得到了这个工作。这是一项正在进行中的工作，我正在添加它，因为这对我来说并且相对简单，所以我没有评论它。 我从这个答案中的代码开始并以此为基础。

Python3 Websockets / Multithreading issue

import asyncio
import websockets
import sys
import time
import multiprocessing

class connect():
    def __init__(self, results, tasks):
        self.x = 0
        self.results = results
        self.tasks = tasks
        self.loop = asyncio.new_event_loop()

    async def commander_start(self):
        while not self.tasks.empty():
            self.uri = self.tasks.get()
            self.tasks.task_done()
            self.ws = await websockets.connect(self.uri)
            while True:
                await asyncio.sleep(0.1)
                print('Listening for a message...')
                while self.x < 5:
                    await self.ws.send("hello")
                    message = await self.ws.recv()
                    message = message+str(self.x)
                    print("< {}".format(message))
                    print('\n\n')
                    print(self.x)
                    self.x += 1
                    self.results.put(message)    
                self.ws.close()
                self.x = 0
                print('ws clsed')
                self.task.cancel()
                await asyncio.sleep(1)
                self.loop.close()


    def run_commander(self):
        self.task = self.loop.create_task(self.commander_start())
        self.loop.run_forever()


    def main(self):
        self.commander = multiprocessing.Process(target=self.run_commander)
        self.commander.start()
        time.sleep(3)
        self.commander.kill()


        print('is alive:', self.commander, self.commander.is_alive())

if __name__ == "__main__":
    size_q = 10

    tasks = multiprocessing.JoinableQueue(maxsize=size_q)
    results = multiprocessing.Queue(maxsize=size_q)
    conn = connect(results,tasks)
    tasks.put('ws://demos.kaazing.com/echo')

    conn.main()
    print('tasks2 put')
    tasks.put('wss://echo.websocket.org')
    conn.main()
    if not results.empty():
        for x in range(size_q):
            print(results.get())

我要更改和改进很多东西，我只是希望基础系统能够正常工作，以便我可以从那里构建，因此任何使用它的人都需要对其进行修改以满足他们的需求。例如，我生成一个新进程并杀死它，而不是运行一个连续进程并让它工作，我还试图找出可加入队列的细节以及如何使用它在进程之后添加作业和事件循环已创建。