V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
推荐学习书目
Learn Python the Hard Way
Python Sites
PyPI - Python Package Index
http://diveintopython.org/toc/index.html
Pocoo
值得关注的项目
PyPy
Celery
Jinja2
Read the Docs
gevent
pyenv
virtualenv
Stackless Python
Beautiful Soup
结巴中文分词
Green Unicorn
Sentry
Shovel
Pyflakes
pytest
Python 编程
pep8 Checker
Styles
PEP 8
Google Python Style Guide
Code Style from The Hitchhiker's Guide
QiShine
V2EX  ›  Python

关于 websockets 异步 IO 的一个菜鸟疑问

  •  
  •   QiShine · 322 天前 · 1913 次点击
    这是一个创建于 322 天前的主题,其中的信息可能已经有所发展或是发生改变。
    对一个 websockets ,在 client 上依次发出
    await ws.send("read A")
    A=await ws.recv()
    await ws.send("read B")
    B=await ws.recv()
    await ws.send("read C")
    C=await ws.recv()
    如果服务器并不是按照请求顺序来回复,比如服务器发出
    Data B
    Data A
    Data C
    那么客户端能保证,A 的内容是"Data A"而不是"Data B" 吗?

    给每个请求和回复一个相同的 ID,但是我不想在 recv 过后,还需要检查 ID,为啥没有 await cecv(ID),只获取指定 ID 的响应吗?
    18 条回复    2023-12-22 15:11:54 +08:00
    ChefIsAwesome
        1
    ChefIsAwesome  
       322 天前
    搞个 id 啊
    Zhuzhuchenyan
        2
    Zhuzhuchenyan  
       322 天前
    第一,websockets 作为一个单纯的协议并不应该关心客户端和服务端双方是如何对具体信息交互的,对于你的需求,唯一的方法就是自己封装一个更上层的解决方案来维护消息自身逻辑的收发有序性。
    第二,从你给的代码,在第一次发送之后,在收到服务器回应之前,第二次信息不会被发送,自然也就不存在你所说的问题,我猜你想描述的是以下这种情况
    await ws.send("read A")
    await ws.send("read B")
    await ws.send("read C")
    # 此时如何保证 A ,B ,C 分别是 Data A ,B ,C
    A=await ws.recv()
    B=await ws.recv()
    C=await ws.recv()
    beyondstars
        3
    beyondstars  
       322 天前
    您好,ws 自身的确无法保证 recv 和 send 的顺序一致。

    ws 是为了解决什么问题呢?它主要是为了解决: 1 ) framing (因为 tcp 连接是一个 stream, tcp 连接是 boundless 的); 2 )连接复用; 3 )双向通信(让 server 也可以主动给 web client 发消息)。这些都跟顺序没关系。

    你需要自己再在 ws 之上实现一层。
    009694
        4
    009694  
       322 天前 via iPhone
    你服务端乱序返回关 ws 协议啥事? ws 接收时本身严格遵循发送顺序 当你应用层本身乱了 就得靠应用层自己排序
    spicy777
        5
    spicy777  
       322 天前 via iPhone
    搞个 id 吧
    QiShine
        6
    QiShine  
    OP
       322 天前
    我就是想问问有没有现成的 await recv(ID),如果没有,那么我想的是建一个 dict {ID: queue of server messages},这样合理吗?有没有更好的办法?
    LinePro
        7
    LinePro  
       322 天前
    你可以自己封装 websocket 协议上层的处理。建一个 dict 存放 id 到 asyncio.Future 的映射。调用处 await 这个 future 。websocket 收到后从 dict 中取出对应的 future 并 set_result 。调用处的 await 就可以得到你设置的 result 内容。
    realJamespond
        8
    realJamespond  
       322 天前
    服务器收到是按顺序的,但你咋保证它返回也按顺序? 我先返回 B 再返回 A 不行?
    leonshaw
        9
    leonshaw  
       322 天前
    read B 和 Data B 有因果关系吗?如果有的话,看你代码 A=await ws.recv() 在 await ws.send("read B") 之前,那么 A 不可能是 Data B
    cloud107202
        10
    cloud107202  
       322 天前
    @QiShine 没有 await recv(ID) 这种. websocket 设计上就是两个方向互不影响的流数据传送。你这个需求也许更适合 http/grpc 做那种传统 req-resp 的 pattern
    ZZ74
        11
    ZZ74  
       322 天前
    无论是否有 recv 你发送是按顺序的 sendA sendB....那写入底层包括网络就是按 AB 顺序写入的,就不会乱序。到客户端网卡的时候也许会乱序,因为网络复杂,但是给你应用程序还是 AB 这个顺序
    Trim21
        12
    Trim21  
       322 天前 via Android
    听起来感觉你想在 websocket 上重新实现 http…
    xiangyuecn
        13
    xiangyuecn  
       322 天前
    简单借鉴 http 的 request response ,一个 response 对应一个 request ,有的请求 1ms 返回响应,有的请求 100 秒返回响应

    请求中简单加一个标记(比如消息编号),服务器返回响应时,原样携带此标记,代表是哪个 request 的 response
    QiShine
        14
    QiShine  
    OP
       321 天前
    @cloud107202 是不是可以有
    dataA = await recv('A')
    doSomethingWith(dataA)
    dataB = await recv('B')
    doAnotherWith(dataB)
    无论 AB 的响应,谁先来,都可以正常运行?像 erlang 里那样通过模式匹配来过滤消息队列
    如果是
    data = await recv()
    if data.ID == A:
    doSomethingWith(dataA)
    elif data.ID == B:
    doAnotherWith(dataB)
    这样的话,肯定可以工作,但是把 A 和 B 的逻辑搅合在一起了,感觉不舒服。
    没啥真实的需求,只是在熟悉 async 和 await 时的一点疑问。
    QiShine
        15
    QiShine  
    OP
       321 天前
    @Zhuzhuchenyan 对,就是你说的这种情况。而且如果有 await ws.send('subscribe D'),后续就会有多次响应,就更复杂了,感觉用 await async 表达异步方式的逻辑,还是不自然。
    dode
        16
    dode  
       321 天前
    服务器加接口,abc 一次性发过去,服务器按顺序处理
    cloud107202
        17
    cloud107202  
       321 天前
    @QiShine 就是你说的后面那种写法,这是业务的固有复杂度,躲不开的。看的不顺眼就抽象个 dispatcher 的逻辑

    // data-flow-in:

    while(true) {
    bytes = await io.recv()
    // 反序列化
    data = decode(bytes)
    // 派发
    xxxhandler.process(data); // 再里面可以 if else
    }
    LinePro
        18
    LinePro  
       321 天前
    之前撸过一份用 websocket 实现 jsonrpc 双向通信的客户端代码,可以参考一下。这里的双向通信是指既支持本地调用远程服务端的 rpc 接口,服务端也可以调用本地客户端提供的 rpc 接口。感觉异步 IO 的关键是要灵活使用 asyncio 提供的异步设施。封装好底层通信之后,上层业务处调用 rpc 就和调用一个普通函数一样简单了。

    https://gist.github.com/linepro6/f51ac8930882ce8200f8a0ae795c214e
    关于   ·   帮助文档   ·   博客   ·   API   ·   FAQ   ·   实用小工具   ·   2497 人在线   最高记录 6679   ·     Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.5 · 24ms · UTC 15:55 · PVG 23:55 · LAX 07:55 · JFK 10:55
    Developed with CodeLauncher
    ♥ Do have faith in what you're doing.