即使只有一个线程在线程池中,并发会发生并发吗？

Question

我正在使用Rails 5和Ruby 2.4.我怎么能弄明白,或者你能看出下面的内容,是否有多个线程在同一时间运行？

pool = Concurrent::FixedThreadPool.new(1)
promises = links.map do |link|
  Concurrent::Promise.execute(executor: pool) do
    result = process_link(link)
    if result
      if result.kind_of?(Array)
        result.each do |my_obj|
          my_obj.update_attributes({ :a => a })
          records_processed = records_processed + my_obj.matches.count
        end
      else
        records_processed = records_processed + result.matches.count
        result.update_attributes({ :a => a })
      end
    end
  end
end
promises.map(&:wait).map(&:value!)

由于我已将我的池设置为"1",我的假设是没有任何内容同时运行,但我不断收到此错误...

Error during processing: (ActiveRecord::ConnectionTimeoutError) could not obtain a connection from the pool within 5.000 seconds (waited 5.002 seconds); all pooled connections were in use
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:202:in `block in wait_poll'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:193:in `loop'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:193:in `wait_poll'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:154:in `internal_poll'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:278:in `internal_poll'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:148:in `block in poll'
/Users/nataliab/.rvm/rubies/ruby-2.4.0/lib/ruby/2.4.0/monitor.rb:214:in `mon_synchronize'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:158:in `synchronize'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:148:in `poll'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:717:in    `acquire_connection'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:490:in `checkout'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:364:in `connection'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:883:in `retrieve_connection'
/Users/nataliab/.rvm/gems/ruby-2.4.0@global/gems/activerecord-5.0.1/lib/active_record/connection_handling.rb:128:in `retrieve_connection'

我没有得到上面的错误,如果我修改我的代码运行我肯定没有并发进行...

links.each do |link|
  result = process_link(link)
  if result
    if result.kind_of?(Array)
      result.each do |race|
        my_obj.update_attributes({ :a => a })
        records_processed = records_processed + my_obj.matches.count
      end
    else
      records_processed = records_processed + result.matches.count
      result.update_attributes({ :a => a })
    end
  end
end

编辑:这是我的开发环境的数据库配置.另请注意,所有这些都在rails控制台中运行.

development:
  adapter: postgresql
  encoding: utf8
  database: sims 
  username: postgres
  password: password
  pool: 5
  timeout: 15000
  host: 127.0.0.1

Answer 1

您假设多个线程必须仅因为连接池耗尽而并发运行是不正确的.仅仅因为连接仍然从连接池"检出"并不意味着当前正在线程中的签出连接上执行查询,这只是意味着线程的连接尚未被检入.线程可以处于空闲状态,但只要尚未明确终止,它仍然可以保持连接池的连接.

由于ActiveRecord Connections是线程本地的,因此您可以通过在多个线程上运行ActiveRecord查询来耗尽连接池,就像在这种情况下一样.(每次Concurrent::FixedThreadPool.new(1)调用时,都会创建一个新线程.)即使您一次只在一个线程上运行查询,默认情况下,每个线程上的连接仍将保持打开状态,直到它们终止.

为避免这种情况,您可以在使用它们之后手动签入连接,或者确保终止(终止)线程,以便池可以恢复(收集)它们的连接.

• 要手动签入连接,请参阅选项ConnectionPool文档.最简单的方法是将ActiveRecord代码包装在一个with_connection块中:

  Concurrent::Promise.execute(executor: pool) do
    ActiveRecord::Base.connection_pool.with_connection do
      # update_attributes, etc
    end
  end
  

• 为了确保所有线程终止,呼叫#shutdown其次#wait_for_termination线程池你使用它完成之后:

  values = promises.map(&:value!)
  pool.shutdown
  pool.wait_for_termination
  

Answer 2

您假设只有一个线程是不正确的。有两个 - 一个在线程池中，一个在线程池中产生一个。

当您让主线程等待并且它不应该访问数据库时，您可能会感到困惑。这并不意味着它仍然没有连接，因此阻止了另一个线程获取连接。

根据经验，您的数据库连接池应该设置为至少产生的线程数+ 1。在这种情况下 - 2。

轻松重现的代码：

# migration
class CreateFoos < ActiveRecord::Migration[5.0]
  def change
    create_table :foos do |t|
      t.integer :bar
    end
  end
end

# model
class Foo < ApplicationRecord
end

# rake task
task experiment: :environment do
  Foo.create
  pool = Concurrent::FixedThreadPool.new(1) 
  promise = 
    Concurrent::Promise.execute(executor: pool) do
      Foo.first.update_attributes!(bar: rand(-42..42))
    end
  promise.wait.value!
end

pool在您的中设置为 1config/database.yml并运行该任务。你会得到一个错误。将其设置为 2 - 就好了。

您可以增加池中的线程数并添加至少那么多的处理承诺。对于数据库连接池=线程池中的线程数，您将始终失败，如果在config/database.yml.