亚洲免费在线-亚洲免费在线播放-亚洲免费在线观看-亚洲免费在线观看视频-亚洲免费在线看-亚洲免费在线视频

Python與Redis的連接教程

系統 1926 0

今天在寫zabbix storm job監控腳本的時候用到了python的redis模塊,之前也有用過,但是沒有過多的了解,今天看了下相關的api和源碼,看到有ConnectionPool的實現,這里簡單說下。
在ConnectionPool之前,如果需要連接redis,我都是用StrictRedis這個類,在源碼中可以看到這個類的具體解釋:
?
redis.StrictRedis Implementation of the Redis protocol.This abstract class provides a Python interface to all Redis commands and an
implementation of the Redis protocol.Connection and Pipeline derive from this, implementing how the commands are sent and received to the Redis server

使用的方法:
?

            
 r=redis.StrictRedis(host=xxxx, port=xxxx, db=xxxx)
 r.xxxx()

          

有了ConnectionPool這個類之后,可以使用如下方法
?

            
pool = redis.ConnectionPool(host=xxx, port=xxx, db=xxxx)
r = redis.Redis(connection_pool=pool)

          

這里Redis是StrictRedis的子類
簡單分析如下:
在StrictRedis類的__init__方法中,可以初始化connection_pool這個參數,其對應的是一個ConnectionPool的對象:
?

            
class StrictRedis(object):
........
  def __init__(self, host='localhost', port=6379,
         db=0, password=None, socket_timeout=None,
         socket_connect_timeout=None,
         socket_keepalive=None, socket_keepalive_options=None,
         connection_pool=None, unix_socket_path=None,
         encoding='utf-8', encoding_errors='strict',
         charset=None, errors=None,
         decode_responses=False, retry_on_timeout=False,
         ssl=False, ssl_keyfile=None, ssl_certfile=None,
         ssl_cert_reqs=None, ssl_ca_certs=None):
     if not connection_pool:
       ..........
       connection_pool = ConnectionPool(**kwargs)
     self.connection_pool = connection_pool

          

在StrictRedis的實例執行具體的命令時會調用execute_command方法,這里可以看到具體實現是從連接池中獲取一個具體的連接,然后執行命令,完成后釋放連接:

?

            
  # COMMAND EXECUTION AND PROTOCOL PARSING
  def execute_command(self, *args, **options):
    "Execute a command and return a parsed response"
    pool = self.connection_pool
    command_name = args[0]
    connection = pool.get_connection(command_name, **options) #調用ConnectionPool.get_connection方法獲取一個連接
    try:
      connection.send_command(*args) #命令執行,這里為Connection.send_command
      return self.parse_response(connection, command_name, **options)
    except (ConnectionError, TimeoutError) as e:
      connection.disconnect()
      if not connection.retry_on_timeout and isinstance(e, TimeoutError):
        raise
      connection.send_command(*args) 
      return self.parse_response(connection, command_name, **options)
    finally:
      pool.release(connection) #調用ConnectionPool.release釋放連接

          

在來看看ConnectionPool類:

            
class ConnectionPool(object): 
    ...........
  def __init__(self, connection_class=Connection, max_connections=None,
         **connection_kwargs):  #類初始化時調用構造函數
    max_connections = max_connections or 2 ** 31
    if not isinstance(max_connections, (int, long)) or max_connections < 0: #判斷輸入的max_connections是否合法
      raise ValueError('"max_connections" must be a positive integer')
    self.connection_class = connection_class #設置對應的參數
    self.connection_kwargs = connection_kwargs
    self.max_connections = max_connections
    self.reset() #初始化ConnectionPool 時的reset操作
  def reset(self):
    self.pid = os.getpid()
    self._created_connections = 0 #已經創建的連接的計數器
    self._available_connections = []  #聲明一個空的數組,用來存放可用的連接
    self._in_use_connections = set() #聲明一個空的集合,用來存放已經在用的連接
    self._check_lock = threading.Lock()
.......
  def get_connection(self, command_name, *keys, **options): #在連接池中獲取連接的方法
    "Get a connection from the pool"
    self._checkpid()
    try:
      connection = self._available_connections.pop() #獲取并刪除代表連接的元素,在第一次獲取connectiong時,因為_available_connections是一個空的數組,
      會直接調用make_connection方法
    except IndexError:
      connection = self.make_connection()
    self._in_use_connections.add(connection)  #向代表正在使用的連接的集合中添加元素
    return connection  
  def make_connection(self): #在_available_connections數組為空時獲取連接調用的方法
    "Create a new connection"
    if self._created_connections >= self.max_connections:  #判斷創建的連接是否已經達到最大限制,max_connections可以通過參數初始化
      raise ConnectionError("Too many connections")
    self._created_connections += 1  #把代表已經創建的連接的數值+1
    return self.connection_class(**self.connection_kwargs)   #返回有效的連接,默認為Connection(**self.connection_kwargs)
  def release(self, connection): #釋放連接,鏈接并沒有斷開,只是存在鏈接池中
    "Releases the connection back to the pool"
    self._checkpid()
    if connection.pid != self.pid:
      return
    self._in_use_connections.remove(connection)  #從集合中刪除元素
    self._available_connections.append(connection) #并添加到_available_connections 的數組中
  def disconnect(self): #斷開所有連接池中的鏈接
    "Disconnects all connections in the pool"
    all_conns = chain(self._available_connections,
             self._in_use_connections)
    for connection in all_conns:
      connection.disconnect()


          

execute_command最終調用的是Connection.send_command方法,關閉鏈接為 Connection.disconnect方法,而Connection類的實現:
?

            
class Connection(object):
  "Manages TCP communication to and from a Redis server"
  def __del__(self):  #對象刪除時的操作,調用disconnect釋放連接
    try:
      self.disconnect()
    except Exception:
      pass

          

核心的鏈接建立方法是通過socket模塊實現:

?
???

            
 def _connect(self):
    err = None
    for res in socket.getaddrinfo(self.host, self.port, 0,
                   socket.SOCK_STREAM):
      family, socktype, proto, canonname, socket_address = res
      sock = None
      try:
        sock = socket.socket(family, socktype, proto)
        # TCP_NODELAY
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        # TCP_KEEPALIVE
        if self.socket_keepalive:  #構造函數中默認 socket_keepalive=False,因此這里默認為短連接
          sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
          for k, v in iteritems(self.socket_keepalive_options):
            sock.setsockopt(socket.SOL_TCP, k, v)
        # set the socket_connect_timeout before we connect
        sock.settimeout(self.socket_connect_timeout) #構造函數中默認socket_connect_timeout=None,即連接為blocking的模式
        # connect
        sock.connect(socket_address)
        # set the socket_timeout now that we're connected
        sock.settimeout(self.socket_timeout) #構造函數中默認socket_timeout=None
        return sock
      except socket.error as _:
        err = _
        if sock is not None:
          sock.close()
.....

          

關閉鏈接的方法:
?

            
  def disconnect(self):
    "Disconnects from the Redis server"
    self._parser.on_disconnect()
    if self._sock is None:
      return
    try:
      self._sock.shutdown(socket.SHUT_RDWR) #先shutdown再close
      self._sock.close()
    except socket.error:
      pass
    self._sock = None

          

???????
可以小結如下
1)默認情況下每創建一個Redis實例都會構造出一個ConnectionPool實例,每一次訪問redis都會從這個連接池得到一個連接,操作完成后會把該連接放回連接池(連接并沒有釋放),可以構造一個統一的ConnectionPool,在創建Redis實例時,可以將該ConnectionPool傳入,那么后續的操作會從給定的ConnectionPool獲得連接,不會再重復創建ConnectionPool。
2)默認情況下沒有設置keepalive和timeout,建立的連接是blocking模式的短連接。
3)不考慮底層tcp的情況下,連接池中的連接會在ConnectionPool.disconnect中統一銷毀。


更多文章、技術交流、商務合作、聯系博主

微信掃碼或搜索:z360901061

微信掃一掃加我為好友

QQ號聯系: 360901061

您的支持是博主寫作最大的動力,如果您喜歡我的文章,感覺我的文章對您有幫助,請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧,狠狠點擊下面給點支持吧,站長非常感激您!手機微信長按不能支付解決辦法:請將微信支付二維碼保存到相冊,切換到微信,然后點擊微信右上角掃一掃功能,選擇支付二維碼完成支付。

【本文對您有幫助就好】

您的支持是博主寫作最大的動力,如果您喜歡我的文章,感覺我的文章對您有幫助,請用微信掃描上面二維碼支持博主2元、5元、10元、自定義金額等您想捐的金額吧,站長會非常 感謝您的哦!!!

發表我的評論
最新評論 總共0條評論
主站蜘蛛池模板: 天天做天天爱天天一爽一毛片 | 久久机热re这里只有精品15 | 日韩精品一区二区三区中文3d | 99爱在线精品视频网站 | 99久久这里只精品麻豆 | 亚洲第一在线 | 最近中文字幕无吗高清视频 | 久久久综合九色合综国产 | 国产无套乱子伦精彩是白视频 | 夜夜躁日日躁狠狠久久 | 免费看91毛片 | 国产成人精品一区二三区2022 | 国语一区| 色多多h| 99热这里只有精品9 99热这里只有精品99 | 阿v视频在线观看免费播放 阿v天堂2017 | 在线视频不卡国产在线视频不卡 | 天天干天天操天天爱 | 国产农村一一级特黄毛片 | 久久久久伊人 | 一级毛片牲交大片 | 天天操天天干天搞天天射 | 一区二区三区在线 | 欧 | 中文字幕精品视频在线观看 | 高清国产天干天干天干不卡顿 | 欧美丰满大乳大屁股毛片 | 九九热视频免费 | 中国特黄特级真人毛片 | 97在线免费观看视频 | 国产日韩美国成人 | 欧美激情视频二区三区 | 偷拍肉窝窝视频在线播放 | 香蕉视频成人在线观看 | 久久无码精品一区二区三区 | 久久天天 | 九九视频在线观看6 | 日日狠狠太爽爽 | jizzjizz中国丝袜美女 | 亚洲视频国产视频 | 一级特黄aaa免费 | 777色狠狠一区二区三区香蕉 |