尽管 ping 通，Websocket 在约 2 小时后获得连接壁橱

Posted 2023-02-16

【中文标题】尽管 ping 通，Websocket 在约 2 小时后获得连接壁橱

【英文标题】：Websocket gets connection closet after ~2h despite pinging

【发布时间】：2018-05-13 10:55:40

【问题描述】：

程序从 gdax websocket 获取数据并将其存储到 sqlite 表中。 该程序在前 2-2,5 小时内运行良好，之后我从调试器收到连接关闭错误。 Witch 很奇怪，因为我相信程序能够在没有调试器帮助的情况下捕获这个错误。

这是我的代码的精简版：

import time
import datetime
import json
import sqlite3
import urllib3
import timeit
import urllib.request
import requests
from websocket import create_connection

conn=sqlite3.connect(":memory:")
c = conn.cursor()  
c.execute("CREATE TABLE `Ticker` (  `ID`    INTEGER PRIMARY KEY AUTOINCREMENT,  `Type`  TEXT,   `Sequence`  INTEGER,    `Product_id`    TEXT,   `Price` INTEGER,    `Open_24h`  INTEGER,    `Volume_24h`    INTEGER,    `Low_24h`   INTEGER,    `High_24h`  INTEGER,    `Volume_30d`    INTEGER,    `Best_bid`  INTEGER,    `Best_ask`  INTEGER,    `side`  TEXT,   `time`  TEXT,   `trade_id`  INTEGER,    `last_size` INTEGER)")
class Websocket():

def __init__(self, wsurl="wss://ws-feed.gdax.com", ws=None, agi1=2, ping_start=0,produkty=['ETH-EUR', 'LTC-EUR', 'BTC-EUR'], newdict=):
    self.wsurl = wsurl
    self.ws = None
    self.agi1=agi1
    self.ping_start=ping_start
    self.produkty=produkty
    self.newdict=newdict
    print(self.wsurl)
    print(self.produkty)

def KonektorWebsocketTicker(self):
    x=0
    b=['type', 'sequence', 'product_id', 'price', 'open_24h', 'volume_24h', 'low_24h', 'high_24h', 'volume_30d', 'best_bid', 'best_ask', 'side', 'time', 'trade_id', 'last_size']
    i=0
    dane=json.loads(self.ws.recv())
    typtranzakcji=dane.get('type',None)      
    if typtranzakcji=='error':                                          #obsługa błędu... mam nadzieje
        print(json.dumps(dane, indent=4, sort_keys=True))
    if typtranzakcji=='ticker':
        while i<len(b):
            a = eval("dane.get('" + b[i] + "', None)")
            self.newdict[b[i]]=a
            i=i+1
        c.execute("INSERT INTO Ticker(type, sequence, product_id, price, open_24h, volume_24h, low_24h, high_24h, volume_30d, best_bid, best_ask, side, time, trade_id, last_size) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (self.newdict['type'], self.newdict['sequence'], self.newdict['product_id'], self.newdict['price'], self.newdict['open_24h'], self.newdict['volume_24h'], self.newdict['low_24h'], self.newdict['high_24h'], self.newdict['volume_30d'], self.newdict['best_bid'], self.newdict['best_ask'], self.newdict['side'], self.newdict['time'], self.newdict['trade_id'], self.newdict['last_size']))

def Polacz(self):
    self.ws=create_connection(self.wsurl)

    if self.agi1 == 0:
        kanaly =None
    elif self.agi1 == 1:
        kanaly= "heartbeat"
    elif self.agi1 == 2:
        kanaly=["ticker"]
    elif self.agi1 == 3:
        kanaly="level2"
    else:
        print('ERROR:WRONG ARGUMENT! (arg1)=', self.agi1)

    print(kanaly)

    if kanaly is None:
        self.ws.send(json.dumps('type': 'subscribe', 'product_ids': self.produkty))
    else:
        self.ws.send(json.dumps('type': 'subscribe', 'product_ids': self.produkty, 'channels': ["name": kanaly, 'product_ids': self.produkty,]))    #wysłanie subskrybcji
    while True:
        try:
            if (time.time() - self.ping_start) >= 20:
                self.ws.ping("ping")
                print('%s ping' % time.ctime())
                self.ping_start = time.time()
        except ValueError as e:
            print(' Error :'.format(time.ctime(), e))
        except Exception as e:
            print(' Error :'.format(time.ctime(), e))
        else:
            print('%s  no ping' % time.ctime())

        dane=json.loads(self.ws.recv())


        if self.agi1==2:
            self.KonektorWebsocketTicker()

        conn.commit()
        time.sleep(1)

class Autotrader():
a='1'
b='1'
if a=="1":
    produkty=["BTC-EUR"]

if b=='1':
    webs=Websocket(produkty=produkty)
    webs.Polacz()  

有人可以向我解释问题出在哪里，因为调试器不是很有帮助，随机更改代码并等待 +2h 以获取有关错误的信息并不是解决此问题的好方法

【参考方案1】：

您打算在哪个交易所进行交易？ 通常在 github 上有一些不错的库供大型交易所使用 :)

我认为问题在于“self.ws.recv()”，它会暂停执行，直到收到消息。这会阻止您到达使您的 websocket 保持活动状态的“self.ws.ping("ping")”。

由于我正在做类似的事情，我还没有具体的解决方案，但引用 Threaded, non-blocking websocket client 似乎应该在单独的线程中运行 keepalive，因此它不会被阻塞：

“总会有一个线程专门用于侦听套接字（在这种情况下，主线程进入 run_forever 内的循环等待消息）。如果您想进行其他操作'将需要另一个线程。"

【讨论】：

老实说我忘记了这个问题。我添加了一个异常处理程序from websocket import create_connection, WebSocketConnectionClosedException，并且在代码的try/except 部分中添加了一个从头开始连接的部分。它不漂亮，我知道我应该使用 asincio 之类的东西，但它可以工作