登录

python Websockets (asyncio ver)强制关闭连接

内容来源于 Stack Overflow,遵循 CC BY-SA 4.0 许可协议进行翻译与使用。IT领域专用引擎提供翻译支持

腾讯云小微IT领域专用引擎提供翻译支持

原文
Stack Overflow用户 修改于2018-06-30
  • 该问题已被编辑
  • 提问者: Stack Overflow用户
  • 提问时间: 2018-06-30 04:57

我正在为python >3.5编写代码。我使用的是Websockets 6.0库,这里是: https://github.com/aaugustin/websockets ,我一直叫它们 asyncio Websockets,因为它们是基于异步的。在我的搜索中,有很多“丢失的连接”,但我正在研究如何取消当前的 ws.recv() 。对 .start() 的调用将创建一个帮助器线程来启动异步事件循环。然后接收函数启动并调用连接函数,websocket ws 被实例化。然后,接收函数工作秋季消息。当我准备停止时,会调用一个.stop()。我期望stop函数能停止等待的 ws.recv() 。然后,在keep_running标志设置为false并运行 ws.close() 的情况下,我期望 ws.recv() 结束,when keep_running循环结束。这不是正在发生的事情。我看到了所有的三个停止,但从来没有 receive stop

command is: stop
Do command is stopped
Stop 1
Stop 2
Stop 3
^CException ignored in: <module 'threading' from '/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py'>
Traceback (most recent call last):
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py", line 1294, in _shutdown
    t.join()
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py", line 1056, in join
    self._wait_for_tstate_lock()
  File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/threading.py", line 1072, in _wait_for_tstate_lock
    elif lock.acquire(block, timeout):
KeyboardInterrupt
(pyalmondplus) Pauls-MBP:pyalmondplus paulenright$ 

参考代码:

import threading
import asyncio
import websockets
import json
class PyAlmondPlus:
    def __init__(self, api_url, event_callback=None):
        self.api_url = api_url
        self.ws = None
        self.loop = asyncio.get_event_loop()
        self.receive_task = None
        self.event_callback = event_callback
        self.keep_running = False
    async def connect(self):
        print("connecting")
        if self.ws is None:
            print("opening socket")
            self.ws = await websockets.connect(self.api_url)
        print(self.ws)
    async def disconnect(self):
    async def send(self, message):
    async def receive(self):
        print("receive started")
        while self.keep_running:
            if self.ws is None:
                await self.connect()
            recv_data = await self.ws.recv()
            print(recv_data)
        print("receive ended")
    def start(self):
        self.keep_running = True
        print("Start 1")
        print("Start 2")
        t = threading.Thread(target=self.start_loop, args=())
        print("Start 3")
        t.start()
        print("Receiver running")
    def start_loop(self):
        print("Loop helper 1")
        policy = asyncio.get_event_loop_policy()
        policy.set_event_loop(policy.new_event_loop())
        self.loop = asyncio.get_event_loop()
        self.loop.set_debug(True)
        asyncio.set_event_loop(self.loop)
        self.loop.run_until_complete(self.receive())
        print("Loop helper 2")
    def stop(self):
        print("Stop 1")
        self.keep_running = False
        print("Stop 2")
        self.ws.close()
        print("Stop 3")
浏览 710 关注 0 得票数 1
  • 得票数为Stack Overflow原文数据
原文
修改于2018-06-30
  • 该回答已被编辑
  • 回答者: Stack Overflow用户
  • 回答时间: 2018-06-30 07:01
得票数 0

我正在研究如何取消当前的ws.recv() ...我看到了所有三个停止,但从来没有接收停止。

您的 receive 协程可能会挂起,等待一些数据到达,因此它无法检查 keep_running 标志。

停止正在运行的协程的简单而强大的方法是 cancel 驱动它的异步 Task 。这将立即取消暂停协程,并生成它正在等待的 CancelledError 。当使用cancel时,你根本不需要 keep_running 标志,异常将自动终止循环。

对.start()的调用创建了一个帮助线程来启动异步事件循环。

这是可行的,但是您并不真的需要为每个 PyAlmondPlus 实例创建一个新的线程和一个全新的事件循环。Asyncio被设计为在单个线程内运行,因此一个事件循环实例可以承载任意数量的协程。

这是一个可能的设计,实现了这两个想法(没有用实际的web套接字测试):

# pre-start a single thread that runs the asyncio event loop
bgloop = asyncio.new_event_loop()
_thread = threading.Thread(target=bgloop.run_forever)
_thread.daemon = True
_thread.start()
class PyAlmondPlus:
    def __init__(self, api_url):
        self.api_url = api_url
        self.ws = None
    async def connect(self):
        if self.ws is None:
            self.ws = await websockets.connect(self.api_url)
    async def receive(self):
        # keep_running is not needed - cancel the task instead
        while True:
            if self.ws is None:
                await self.connect()
            recv_data = await self.ws.recv()
    async def init_receive_task(self):
        self.receive_task = bgloop.create_task(self.receive())
    def start(self):
        # use run_coroutine_threadsafe to safely submit a coroutine
        # to the event loop running in a different thread
        init_done = asyncio.run_coroutine_threadsafe(
            self.init_receive_task(), bgloop)
        # wait for the init coroutine to actually finish
        init_done.result()