2

我有一个 Flask Web 应用程序，我想在其中保持与 AWS Neptune 图形数据库的持久连接。该连接建立如下：

from gremlin_python.process.anonymous_traversal import traversal
from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection

neptune_endpt = 'db-instance-x.xxxxxxxxxx.xx-xxxxx-x.neptune.amazonaws.com'
remoteConn = DriverRemoteConnection(f'wss://{neptune_endpt}:8182/gremlin','g')
self.g = traversal().withRemote(remoteConn)

我面临的问题是，如果闲置，连接会自动断开，并且我找不到检测连接是否断开的方法（以便我可以使用上面的代码片段重新连接）。

我见过类似的问题：Gremlin server withRemote connection closed - how to reconnect automatic? 但是这个问题也没有解决方案。这个类似的问题也没有答案。

我尝试了以下两种解决方案（均无效）：

  1. 我在四个 Gunicorn 工作人员后面设置了我的 web 应用程序，超时时间为 100 秒，希望工作人员重新启动能够处理 Gremlin 超时。
  2. 我尝试捕获异常以检测连接是否断开。每次我self.g在我的图表上进行一些遍历时，我都会尝试“刷新”连接，我的意思是：
def _refresh_neptune(self):
    try:
        self.g = traversal().withRemote(self.conn)
    except:
        self.conn = DriverRemoteConnection(f'wss://{neptune_endpt}:8182/gremlin','g')
        self.g = traversal().withRemote(self.conn)

这里self.conn被初始化为：

self.conn = DriverRemoteConnection(f'wss://{neptune_endpt}:8182/gremlin','g')

有没有办法解决这个连接错误？

谢谢

更新：添加了以下错误消息：

  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/process/traversal.py
", line 58, in toList
    return list(iter(self))
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/process/traversal.py
", line 48, in __next__
    self.traversal_strategies.apply_strategies(self)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/process/traversal.py
", line 573, in apply_strategies
    traversal_strategy.apply(traversal)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/remote_connec
tion.py", line 149, in apply
    remote_traversal = self.remote_connection.submit(traversal.bytecode)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/driver_remote
_connection.py", line 56, in submit
    results = result_set.all().result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 425, in result
    return self.__get_result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/resultset.py"
, line 90, in cb
    f.result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 425, in result
    return self.__get_result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/usr/lib/python3.6/concurrent/futures/thread.py", line 56, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/connection.py
", line 83, in _receive
    status_code = self._protocol.data_received(data, self._results)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/protocol.py",
 line 81, in data_received
    'message': 'Server disconnected - please try to reconnect', 'attributes': {}})
gremlin_python.driver.protocol.GremlinServerError: 500: Server disconnected - please try to reconnect
4

1 回答 1

1

我不确定这是解决这个问题的最佳方法，但我也在使用 gremlin-python 和 Neptune，我也遇到了同样的问题。我通过实现可以提供给 DriverRemoteConnection 的传输来解决它。

DriverRemoteConnection(
    url=endpoint,
    traversal_source=self._traversal_source,
    transport_factory=Transport
)

gremlin-python 在异常时返回连接池，当连接关闭时返回的异常是 GremlinServerError ，它也会因其他错误而引发。

gremlin_python/driver/connection.py#L69 - gremlin_python/driver/protocol.py#L80

自定义传输与 gremlin-python 的 TornadoTransport 相同，但读写方法扩展为：

  • 如果 Web 套接字客户端已关闭，则重新打开已关闭的连接
  • 如果 Web 套接字客户端从 read_message 返回 None，则引发 StreamClosedError

添加回池的死连接可以重新打开，然后您可以处理 StreamClosedError 以应用一些重试逻辑。我通过覆盖 DriverRemoteConnection 中的 submit 和 submitAsync 方法来做到这一点，但是您可以在任何地方捕获并重试。

class Transport(AbstractBaseTransport):
    def __init__(self):
        self._ws = None
        self._loop = ioloop.IOLoop(make_current=False)
        self._url = None

        # Because the transport will try to reopen the underlying ws connection
        # track if the closed() method has been called to prevent the transport
        # from reopening.
        self._explicit_closed = True

    @property
    def closed(self):
        return not self._ws.protocol

    def connect(self, url, headers=None):
        self._explicit_closed = False

        # Set the endpoint URL
        self._url = httpclient.HTTPRequest(url, headers=headers) if headers else url

        # Open the connection
        self._connect()

    def write(self, message):
        # Before writing, try to ensure that the connection is open.
        if self.closed:
            self._connect()

        self._loop.run_sync(lambda: self._ws.write_message(message, binary=True))

    def read(self):
        result = self._loop.run_sync(self._ws.read_message)

        # If the read call returns None, the stream has closed.
        if result is None:
            self._ws.close()  # Ensure we close the stream
            raise StreamClosedError()

        return result

    def close(self):
        self._ws.close()
        self._loop.close()
        self._explicit_closed = True

    def _connect(self):
        # If close() was called explicitly on the transport, don't allow
        # subsequent calls to write() to reopen the connection.
        if self._explicit_closed:
            raise TransportClosedError(
                "Transport has been closed and can not be reopened."
            )

        # Check if the ws is closed, if it is not, close it.
        if self._ws and not self.closed:
            self._ws.close()

        # Open the ws connection
        self._ws = self._loop.run_sync(
            lambda: websocket.websocket_connect(url=self._url)
        )


class TransportClosedError(Exception):
    pass

这也适用于 gremlin-pythons 连接池。

如果您不需要池化，另一种方法是将池大小设置为 1 并实现某种形式的 keep-alive，就像这里讨论的那样。小叮当-2352

看起来 gremlin-python 中的网络套接字 ping/keep-alive 尚未实现TINKERPOP-1886。

于 2020-08-24T23:13:14.860 回答