---

title: Python 读写 hbase 数据的正确姿势（五）

tags:

• hbase
• happybase
• python

categories:

• �Hbase

comments: true
date: 2017-10-31 23:00:00

---

为什么一条异常的连接会出现在 connect pool 中，而且总会拿到这条连接 ？

在本系列第三篇文章末尾提出了这样一个问题 『 为什么一条异常的连接会出现在 connect pool 中，而且总会拿到这条连接 』，本文将继续深入这个问题，找到其根本原因。

问题分析

弄清这个问题，要深入 happybase 的代码看一下具体实现逻辑。

happybase 的连接池要求必须要使用 context manager，以下是其通过 context manager 获取/归还 Connection 的源码：

    @contextlib.contextmanager
    def connection(self, timeout=None):
        # 获取一个 Connection 对象
        connection = getattr(self._thread_connections, 'current', None)
        return_after_use = False
        if connection is None：
            return_after_use = True
            connection = self._acquire_connection(timeout)
            with self._lock:
                self._thread_connections.current = connection

        try:
            # 打开连接，并返回一个可用的连接给contextmanager
            connection.open()
            yield connection
        except (TException, socket.error):
            # 捕获 Trhift 和 socket 的异常和错误，则需要 refresh thrift client，保证最后归还的 Connection 的 thrift client 是可用的。
            logger.info("Replacing tainted pool connection")
            connection._refresh_thrift_client()
            connection.open()
            raise
        finally:
            # �最终会归还这条连接
            if return_after_use:
                del self._thread_connections.current
                self._return_connection(connection)

实现逻辑大概可以归纳为：

• 在 pool 初始化的时候构建一个包含若干 Connection 的 queue 以及一个 lock。
• 当向连接池请求连接时，open 一个 Connection 并返回，在这个过程中会 catch TException 以及 socket.error。
• 如果 catch 到以上 error/exception 则会刷新 Connection 对象的thrift_client，保证在退出 context manager 时返回给队列的 Connection 是可用的。
• 退出 context manager 前将 Connection 返还。

从以上逻辑可以发现，happybase 是通过 context manager 保证 Connection 在退出时是正常的，而我们的场景中出现了 socket.error 却并没有被 catch 住，说明有可能是错误发生在 context manager 之外，回到代码：

def recent_events_v1(start, end, table=None, filter_str=None, limit=2000):
    with get_connetion_pool().connection() as conn:
        if table is not None:
            t = conn.table(table)
        else:
            t = conn.table(TABLE)
        start_row = 'ARTICLE' + str(start * 1000000)
        end_row = 'ARTICLE' + str(end * 1000000)
        return t.scan(row_start=start_row, row_stop=end_row, filter=filter_str, limit=limit)

从代码可以发现，再使用连接池的过程中，退出 context manager 前直接 return table.scan() ，而 scan 方法会创建一个 scanner ，最终返回一个 generator，到这里基本可以说是水落石出了！

问题原因

因为 generator 的特性，在退出 context manager 前并没有发生真正的查询，所以这时返回给 connent pool 的 Connection 仍然是没有问题的。只用在真正遍历这个 generator 时才会发生数据查询，而这个过程肯定在 context manager 之外，所以此时如果出现 error 则不会再有类似的 catch 逻辑去保证这条 Connection 在发生异常时去刷新 thrift_client，最终导致这条已经归还给 pool 的 Connection 失效了。

同时这里还会存在另一个问题，在并发场景下，遍历 generator 发生查询时可能这个 Connection 已经被分配给其他线程使用了，导致这个 Connection 同时被两个线程所有，出现一些难以预测的问题。

解决问题

既然发现了问题的根因，解决起来就比较简单了，只要保证所有使用 Connection 的逻辑都发生在 context manager 内就好，所以这里可以把遍历generator 的逻辑放在 context manager 内，最终返回一个 list 对象而不是 generator，代码如下：

def recent_events_v4(start, end, table=None, filter_str=None, limit=2000):
    with get_connetion_pool().connection() as conn:
        if table is not None:
            t = conn.table(table)
        else:
            t = conn.table(TABLE)
        start_row = 'ARTICLE' + str(start * 1000000)
        end_row = 'ARTICLE' + str(end * 1000000)
        result_generator = t.scan(row_start=start_row, row_stop=end_row, filter=filter_str, limit=limit)
        return [l for l in result_generator]

从下图可以看出，修正后的代码在运行的过程中出现 TTransportException后，不会在出现 Broken pipe。

不会在出现 Broken pipe

继续思考

上面分析了『为什么一条异常的连接会出现在 connect pool 中』，那 『而且总会拿到这条连接 』 又是为什么?

仔细观察 happybase 获取 connection 的前半部分逻辑可以发现，其是优先从self._thread_connections 获取链接对象，当获取不到时才通过 self._acquire_connection 从 pool 中取。

这个 self._thread_connections 是个什么东西？这是一个 thread local 变量，即线程自有的局部变量，其他线程不可访问，happybase 源码：

self._thread_connections = threading.local()

某个线程第一次请求获取 Connection 时，通过这个 thread local 变量，把分配给它的 Connection 记录下来，当下次这个线程再请求时，则优先把这个 thread local 变量记录的 Connection 返回给线程。

因为我们是单线程场景，所以每次返回给主进程的都是同一个有问题的 Connection 对象，这就解释了 『总会拿到问题链接』 这个问题。

小结

综上，在使用连接池的场景中，注意类似 scan 这种具有延时行的操作，一定要放在 context mananer 内，才能保证连接池内的连接可用~