类 Bundler::ConnectionPool

用于在多个线程之间共享有限数量的对象或网络连接的通用连接池类。注意：池元素是延迟创建的。

使用块的示例用法（更快）

@pool = Bundler::ConnectionPool.new { Redis.new }
@pool.with do |redis|
  redis.lpop('my-list') if redis.llen('my-list') > 0
end

使用可选的超时覆盖（用于单次调用）

@pool.with(timeout: 2.0) do |redis|
  redis.lpop('my-list') if redis.llen('my-list') > 0
end

替换现有连接的示例用法（较慢）

$redis = Bundler::ConnectionPool.wrap { Redis.new }

def do_work
  $redis.lpop('my-list') if $redis.llen('my-list') > 0
end

接受以下选项

:size - 池中的连接数，默认为 5
:timeout - 如果当前没有可用连接，等待连接的时间，默认为 5 秒
:auto_reload_after_fork - fork 后自动删除所有连接，默认为 true

常量

DEFAULTS
INSTANCES
VERSION

属性

auto_reload_after_fork[R]

fork 后自动删除所有连接

size[R]

此连接池的大小

公共类方法

after_fork() 点击以切换源代码

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 52
def self.after_fork
  INSTANCES.values.each do |pool|
    next unless pool.auto_reload_after_fork

    # We're on after fork, so we know all other threads are dead.
    # All we need to do is to ensure the main thread doesn't have a
    # checked out connection
    pool.checkin(force: true)
    pool.reload do |connection|
      # Unfortunately we don't know what method to call to close the connection,
      # so we try the most common one.
      connection.close if connection.respond_to?(:close)
    end
  end
  nil
end

new(options = {}, &block) 点击以切换源代码

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 90
def initialize(options = {}, &block)
  raise ArgumentError, "Connection pool requires a block" unless block

  options = DEFAULTS.merge(options)

  @size = Integer(options.fetch(:size))
  @timeout = options.fetch(:timeout)
  @auto_reload_after_fork = options.fetch(:auto_reload_after_fork)

  @available = TimedStack.new(@size, &block)
  @key = :"pool-#{@available.object_id}"
  @key_count = :"pool-#{@available.object_id}-count"
  INSTANCES[self] = self if INSTANCES
end

wrap(options, &block) 点击以切换源代码

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 44
def self.wrap(options, &block)
  Wrapper.new(options, &block)
end

公共实例方法

available() 点击以切换源代码

此刻可用于签出的池条目数量。

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 169
def available
  @available.length
end

checkin(force: false) 点击以切换源代码

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 129
def checkin(force: false)
  if ::Thread.current[@key]
    if ::Thread.current[@key_count] == 1 || force
      @available.push(::Thread.current[@key])
      ::Thread.current[@key] = nil
      ::Thread.current[@key_count] = nil
    else
      ::Thread.current[@key_count] -= 1
    end
  elsif !force
    raise Bundler::ConnectionPool::Error, "no connections are checked out"
  end

  nil
end

checkout(options = {}) 点击以切换源代码

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 119
def checkout(options = {})
  if ::Thread.current[@key]
    ::Thread.current[@key_count] += 1
    ::Thread.current[@key]
  else
    ::Thread.current[@key_count] = 1
    ::Thread.current[@key] = @available.pop(options[:timeout] || @timeout)
  end
end

reload(&block) 点击以切换源代码

通过将每个连接传递给 block，然后从池中删除它来重新加载 Bundler::ConnectionPool。后续签出会根据需要创建新连接。

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 159
def reload(&block)
  @available.shutdown(reload: true, &block)
end

shutdown(&block) 点击以切换源代码

通过将每个连接传递给 block，然后从池中删除它来关闭 Bundler::ConnectionPool。在关闭后尝试签出连接将引发 Bundler::ConnectionPool::PoolShuttingDownError。

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 150
def shutdown(&block)
  @available.shutdown(&block)
end

then(options = {})

别名为：with

with(options = {}) { |conn| ... } 点击以切换源代码

# File bundler/vendor/connection_pool/lib/connection_pool.rb, line 105
def with(options = {})
  Thread.handle_interrupt(Exception => :never) do
    conn = checkout(options)
    begin
      Thread.handle_interrupt(Exception => :immediate) do
        yield conn
      end
    ensure
      checkin
    end
  end
end

也别名为：then