通过 UDS 发送大 JSON 数据

Question

我正在尝试通过 UDS 套接字和 UDP 协议将长 json 字符串从一个进程发送到另一个进程，如下所示：

# server.py
def main():
    if os.path.exists(UDSFILE):
        os.remove(UDSFILE)

    sock = socket(AF_UNIX, SOCK_DGRAM)
    sock.bind(UDSFILE)

    file = open(CSVFILE, 'ab')
    msg = b''
    while True:
        while True:
            try:
                package, *_ = sock.recvfrom(4096)
                msg += package
            except ConnectionError:
                log.exception()
                raise
            if b'\t' in msg:
                data, msg = msg.split(b'\t', 1)
                break
        data = json.loads(data.decode())
        exchange, ts, data = data
        file.write(("%s\t%s\t%s\n" % (exchange, ts, data)).encode())

# Client.py

def main():
    wss = GeminiWSS(args.pair)
    wss.start()

    sock = socket(AF_UNIX, SOCK_DGRAM)

    while True:
        try:
            sock.sendto((json.dumps(wss.get()) + '\t').encode(), socket_addr.encode())
        except Exception as e:
            log.exception(e)
            log.info("Attempting to restart..")
            wss.restart()
        except OSError as e:
            log.exception(e)
            print("OSError on connecting, trying again..")

服务器在解码时崩溃，跟踪如下：

Traceback (most recent call last):
  File "scribe.py", line 62, in <module>
    main()
  File "scribe.py", line 49, in main
    data = json.loads(data.decode())
  File "/opt/anaconda3/lib/python3.5/json/__init__.py", line 319, in loads
    return _default_decoder.decode(s)
  File "/opt/anaconda3/lib/python3.5/json/decoder.py", line 339, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/opt/anaconda3/lib/python3.5/json/decoder.py", line 355, in raw_decode
    obj, end = self.scan_once(s, idx)
json.decoder.JSONDecodeError: Expecting ':' delimiter: line 1 column 4099 (char 4098)

如您所见，我尝试使用自定义分隔符来读取接收到的数据 (\t)，但这不起作用（因此 JSONDecodeError）。

那么，我如何才能确保我的数据正确组装？不幸的是，简单地增加缓冲区是无效的，因为我没有我所知道的最大消息长度。

编辑代码生成从 wss.get():

返回的数据

def _subscription_thread(self):
    """
    Thread Method, running the connection for each endpoint.
    :param endpoint:
    :return:
    """
    try:
        conn = create_connection(self.addr + self.endpoint, timeout=5)
    except WebSocketTimeoutException:
        self._controller_q.put('restart')
        return

    while self.endpoint_thread_running:
        try:
            msg = conn.recv()
        except WebSocketTimeoutException:
            log.exception()
            raise

        log.debug("_subscription_thread(): Putting data on q..")
        try:
            self.data_q.put(('Gemini', json.loads(msg), time.time()), timeout=1)
        except TimeoutError:
            continue
        finally:
            log.debug("_subscription_thread(): Data Processed, looping back..")
    conn.close()
    log.debug("_subscription_thread(): Thread Loop Ended.")

数据样本

导致 'JSONDecodeError 引发的 bytes 对象：

b'["Gemini", {"eventId": 609771498, "events": [{"remaining": "21914.3", "reason": "initial", "price": "0.01", "side": "bid", "type": "change", "delta": "21914.3"}, {"remaining": "0.000047", "reason": "initial", "price": "0.06", "side": "bid", "type": "change", "delta": "0.000047"}, {"remaining": "110", "reason": "initial", "price": "0.10", "side": "bid", "type": "change", "delta": "110"}, {"remaining": "1053.33333333", "reason": "initial", "price": "0.15", "side": "bid", "type": "change", "delta": "1053.33333333"}, {"remaining": "1", "reason": "initial", "price": "0.90", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "744", "reason": "initial", "price": "1.00", "side": "bid", "type": "change", "delta": "744"}, {"remaining": "2400", "reason": "initial", "price": "1.10", "side": "bid", "type": "change", "delta": "2400"}, {"remaining": "1", "reason": "initial", "price": "5.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "1", "reason": "initial", "price": "8.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "0.022", "reason": "initial", "price": "9.00", "side": "bid", "type": "change", "delta": "0.022"}, {"remaining": "1", "reason": "initial", "price": "11.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "30", "reason": "initial", "price": "16.00", "side": "bid", "type": "change", "delta": "30"}, {"remaining": "2", "reason": "initial", "price": "24.00", "side": "bid", "type": "change", "delta": "2"}, {"remaining": "400", "reason": "initial", "price": "26.00", "side": "bid", "type": "change", "delta": "400"}, {"remaining": "0.03015", "reason": "initial", "price": "30.00", "side": "bid", "type": "change", "delta": "0.03015"}, {"remaining": "0.97", "reason": "initial", "price": "31.85", "side": "bid", "type": "change", "delta": "0.97"}, {"remaining": "0.029", "reason": "initial", "price": "33.85", "side": "bid", "type": "change", "delta": "0.029"}, {"remaining": "1", "reason": "initial", "price": "36.97", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "2.9975", "reason": "initial", "price": "40.00", "side": "bid", "type": "change", "delta": "2.9975"}, {"remaining": "0.04460443", "reason": "initial", "price": "67.00", "side": "bid", "type": "change", "delta": "0.04460443"}, {"remaining": "16.20200501", "reason": "initial", "price": "99.75", "side": "bid", "type": "change", "delta": "16.20200501"}, {"remaining": "62.08649948", "reason": "initial", "price": "100.00", "side": "bid", "type": "change", "delta": "62.08649948"}, {"remaining": "1", "reason": "initial", "price": "151.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "1", "reason": "initial", "price": "159.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "2.60664747", "reason": "initial", "price": "191.05", "side": "bid", "type": "change", "delta": "2.60664747"}, {"remaining": "1", "reason": "initial", "price": "200.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "0.03976369", "reason": "initial", "price": "220.05", "side": "bid", "type": "change", "delta": "0.03976369"}, {"remaining": "2.6635", "reason": "initial", "price": "300.00", "side": "bid", "type": "change", "delta": "2.6635"}, {"remaining": "0.855", "reason": "initial", "price": "350.00", "side": "bid", "type": "change", "delta": "0.855"}, {"remaining": "0.04172229", "reason": "initial", "price": "359.52", "side": "bid", "type": "change", "delta": "0.04172229"}, {"remaining": "0.02773771", "reason": "initial", "price": "360.52", "side": "bid", "type": "change", "delta": "0.02773771"}, {"remaining": "2.59220779", "reason": "initial", "price": "385.00", "side": "bid", "type": "change", "delta": "2.59220779"}, {"remaining": "3.740625", "reason": "initial", "price": "400.00", "side": "bid", "type": "change", "delta": "3.740625"}, {"remaining": "1.21646341", "reason": "initial", "price": "410.00", "side": "bid", "type": "change", "delta": "1.21646341"}, {"remaining": "1.20471014", "reason": "initial", "price": "414.00", "side": "bid", "type": "change", "delta": "1.20471014"}, {"remai["Gemini", {"eventId": 609771500, "events": [{"remaining": "0", "reason": "cancel", "price": "2053.62", "side": "ask", "type": "change", "delta": "-2"}], "type": "update"}, 1495312354.6720355]'

从此来源生成：

pastebin（移动到 pastebin，由于文本限制）

Answer 1

这与 UDS 无关，但数据本身无效 JSON 字符串。

你的错误信息是这样说的：

json.decoder.JSONDecodeError: Expecting ':' delimiter: line 1 column 4099 (char 4098)

这意味着 json.loads 失败并出现 JSON 解码错误，因为输入数据不是有效的 JSON 字符串。原因指定为

Expecting ':' delimiter

所以你知道数据不对，但在哪里？答案：

line 1 column 4099 (char 4098)

现在让我们获取输入

>>> input_data = b'["Gemini", {"eventId": 609771498, "events": [{"remaining": "21914.3", "reason": "initial", "price": "0.01", "side": "bid", "type": "change", "delta": "21914.3"}, {"remaining": "0.000047", "reason": "initial", "price": "0.06", "side": "bid", "type": "change", "delta": "0.000047"}, {"remaining": "110", "reason": "initial", "price": "0.10", "side": "bid", "type": "change", "delta": "110"}, {"remaining": "1053.33333333", "reason": "initial", "price": "0.15", "side": "bid", "type": "change", "delta": "1053.33333333"}, {"remaining": "1", "reason": "initial", "price": "0.90", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "744", "reason": "initial", "price": "1.00", "side": "bid", "type": "change", "delta": "744"}, {"remaining": "2400", "reason": "initial", "price": "1.10", "side": "bid", "type": "change", "delta": "2400"}, {"remaining": "1", "reason": "initial", "price": "5.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "1", "reason": "initial", "price": "8.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "0.022", "reason": "initial", "price": "9.00", "side": "bid", "type": "change", "delta": "0.022"}, {"remaining": "1", "reason": "initial", "price": "11.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "30", "reason": "initial", "price": "16.00", "side": "bid", "type": "change", "delta": "30"}, {"remaining": "2", "reason": "initial", "price": "24.00", "side": "bid", "type": "change", "delta": "2"}, {"remaining": "400", "reason": "initial", "price": "26.00", "side": "bid", "type": "change", "delta": "400"}, {"remaining": "0.03015", "reason": "initial", "price": "30.00", "side": "bid", "type": "change", "delta": "0.03015"}, {"remaining": "0.97", "reason": "initial", "price": "31.85", "side": "bid", "type": "change", "delta": "0.97"}, {"remaining": "0.029", "reason": "initial", "price": "33.85", "side": "bid", "type": "change", "delta": "0.029"}, {"remaining": "1", "reason": "initial", "price": "36.97", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "2.9975", "reason": "initial", "price": "40.00", "side": "bid", "type": "change", "delta": "2.9975"}, {"remaining": "0.04460443", "reason": "initial", "price": "67.00", "side": "bid", "type": "change", "delta": "0.04460443"}, {"remaining": "16.20200501", "reason": "initial", "price": "99.75", "side": "bid", "type": "change", "delta": "16.20200501"}, {"remaining": "62.08649948", "reason": "initial", "price": "100.00", "side": "bid", "type": "change", "delta": "62.08649948"}, {"remaining": "1", "reason": "initial", "price": "151.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "1", "reason": "initial", "price": "159.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "2.60664747", "reason": "initial", "price": "191.05", "side": "bid", "type": "change", "delta": "2.60664747"}, {"remaining": "1", "reason": "initial", "price": "200.00", "side": "bid", "type": "change", "delta": "1"}, {"remaining": "0.03976369", "reason": "initial", "price": "220.05", "side": "bid", "type": "change", "delta": "0.03976369"}, {"remaining": "2.6635", "reason": "initial", "price": "300.00", "side": "bid", "type": "change", "delta": "2.6635"}, {"remaining": "0.855", "reason": "initial", "price": "350.00", "side": "bid", "type": "change", "delta": "0.855"}, {"remaining": "0.04172229", "reason": "initial", "price": "359.52", "side": "bid", "type": "change", "delta": "0.04172229"}, {"remaining": "0.02773771", "reason": "initial", "price": "360.52", "side": "bid", "type": "change", "delta": "0.02773771"}, {"remaining": "2.59220779", "reason": "initial", "price": "385.00", "side": "bid", "type": "change", "delta": "2.59220779"}, {"remaining": "3.740625", "reason": "initial", "price": "400.00", "side": "bid", "type": "change", "delta": "3.740625"}, {"remaining": "1.21646341", "reason": "initial", "price": "410.00", "side": "bid", "type": "change", "delta": "1.21646341"}, {"remaining": "1.20471014", "reason": "initial", "price": "414.00", "side": "bid", "type": "change", "delta": "1.20471014"}, {"remai["Gemini", {"eventId": 609771500, "events": [{"remaining": "0", "reason": "cancel", "price": "2053.62", "side": "ask", "type": "change", "delta": "-2"}], "type": "update"}, 1495312354.6720355]'

>>> print(input_data[4060:4140])
b'ge", "delta": "1.20471014"}, {"remai["Gemini", {"eventId": 609771500, "events": '

现在清楚了，您的数据在这部分已损坏：

{"remai["Gemini", {"eventId"

现在您需要弄清楚为什么您的数据无效 json。

Answer 2

最终，我不再尝试自己处理 UDS 连接和发送，而是让 python 使用 multiprocessing 库的 Client 和 Listener 类，并提供文件路径。这现在可以正常工作了。

> listener.py
from multiprocessing.connection import Listener

serv = Listener('/path/to/uds')
while True:
    conn = serv.accept()

    try:
        while True:
            msg = conn.recv()
            print(msg)
            conn.send(msg.upper())
    except EOFError:
        print('Conn closed!')
    finally:
        conn.close()  
serv.close()


> client.py
from multiprocessing.connection import Client
large_data = ''.join(['a' for i in range(1000000)])
sock = Client('/path/to/uds')
sock.send(large_data)
answer = sock.recv()
print(answer)
sock.close()

更不用说我无缘无故地拆分了消息，因为 UDS 连接是文件描述符。

通过 UDS 发送大 JSON 数据

Sending larg JSON data via UDS

python

sockets

unix-socket

python-3.x