Rails 数据库异步查询探索

count = 10,000,000
ActiveRecord::Base.connection.execute(<<~SQL)
  TRUNCATE orders;
  INSERT INTO orders (user_id, product_id, quantity, status, created_at, updated_at)
  SELECT (1 + random()*1000)::int,
          (1 + random()*100)::int,
          (1 + random()*10)::int,
          (random()*4)::int,
          now() - ((random()*730)::int || ' days')::interval,
          now()
  FROM generate_series(1, #{count});
  ANALYZE orders;
SQL

config.active_record.async_query_executor = :global_thread_pool # 进程内共享一个线程池
config.active_record.global_executor_concurrency = 4   # 最大并发线程数，默认就是 4

def get_purchase_info_with_load_async
  start_time = Time.now

  # 一口气调度 4 条查询，互不等待；返回 ActiveRecord::Promise
  count_promise = Order.paid.async_count
  sum_promise   = Order.paid.async_sum(:quantity)
  avg_promise   = Order.paid.async_average(:quantity)
  top10_promise = Order.paid.group(:product_id).order(Arel.sql("SUM(quantity) DESC")).limit(10)
                         .async_pluck(:product_id, Arel.sql("SUM(quantity)"))

  # 依次取结果；首次 .value 会阻塞到该查询完成
  total_orders_count    = count_promise.value
  total_sales_quantity  = sum_promise.value
  avg_sales_quantity    = avg_promise.value
  top_10_sales_products = top10_promise.value

  puts "load_async 完成，总耗时 #{((Time.now - start_time) * 1000).round(2)} 毫秒"
end

# Rails + Puma 下，每个 HTTP request/Job 都有一层 ActionDispatch::Executor 包起来
# load_async 需要在 Executor 内执行，Rails console 中需要手动添加 Executor
Rails.application.executor.wrap do
# uncached 避免 Rails 缓存，影响测试结果
  ActiveRecord::Base.uncached { get_purchase_info_with_load_async }
end

# activerecord-8.0.5.1/lib/active_record.rb:286-294
def self.global_thread_pool_async_query_executor # :nodoc:
  concurrency = global_executor_concurrency || 4
  @global_thread_pool_async_query_executor ||= Concurrent::ThreadPoolExecutor.new(
    min_threads: 0,
    max_threads: concurrency,
    max_queue: concurrency * 4,
    fallback_policy: :caller_runs
  )
end

# activerecord-8.0.5.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:700-703
def schedule_query(future_result) # :nodoc:
  @async_executor.post { future_result.execute_or_skip }
  Thread.pass
end

# `@async_executor.post` 调用链：
# @async_executor.post                              concurrent-ruby/concurrent/executor/ruby_executor_service.rb:17-31
# └─ Concurrent::RubyThreadPoolExecutor#ns_execute  concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb:178-187
#    └─ Concurrent::RubyThreadPoolExecutor#ns_assign_worker
#       └─ Concurrent::RubyThreadPoolExecutor#ns_add_busy_worker
#          └─ Worker.new
#             └─ Worker#create_worker

# concurrent-ruby-1.3.8/lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb:356-379
def create_worker(queue, pool, idletime)
  Thread.new(queue, pool, idletime) do |my_queue, my_pool, my_idletime|
  ...

# activerecord-8.0.5.1/lib/active_record/future_result.rb:100-106
def execute_or_skip
  return unless pending?

  @session.synchronize do
    return unless pending?

    @pool.with_connection do |connection|
    ...

# fiber 粒度：每个 fiber 可以有自己的 db 连接。
config.active_support.isolation_level = :fiber

require "async"
require "async/semaphore"
require "async/barrier"

def get_purchase_info_concurrently
  start_time = Time.now
  semaphore = Async::Semaphore.new(4)   # 一个线程内最多 4 个 fiber 同时查库
  barrier   = Async::Barrier.new
  results   = {}

  Async do
    {
      count: -> { Order.paid.count },
      sum:   -> { Order.paid.sum(:quantity) },
      avg:   -> { Order.paid.average(:quantity) },
      top10: -> { Order.paid.group(:product_id).order(Arel.sql("SUM(quantity) DESC")).limit(10)
                         .pluck(:product_id, Arel.sql("SUM(quantity)")).to_h }
    }.each do |label, query|
      barrier.async do
        semaphore.acquire do
          results[label] = query.call
        end
      end
    end
    barrier.wait
  end

  puts "并发完成,总耗时 #{((Time.now - start_time) * 1000).round(2)} 毫秒"
  results
end

ActiveRecord::Base.uncached { get_purchase_info_concurrently }

# lib/async/task.rb:433
def schedule(&block)
  @fiber = Fiber.new(annotation: self.annotation) do
  ...

# async-2.21.1/lib/kernel/async.rb:24-39
def Async(...)
  ...
  # This calls Fiber.set_scheduler(self):
  reactor = ::Async::Reactor.new
  begin
    return reactor.run(...)
  ...

# async-2.21.1/lib/async/scheduler.rb:492-500   阻塞当前线程跑事件循环
def run(...)
  initial_task = self.async(...) if block_given?
  self.run_loop { run_once }
  ...

# async-2.21.1/lib/async/scheduler.rb:272
# Wait for the specified IO to become ready for the specified events.
def io_wait(io, events, timeout = nil)
  ...

# activesupport-8.0.5.1/lib/active_support/isolated_execution_state.rb:24-28
module ActiveSupport
  module IsolatedExecutionState # :nodoc:
    class << self
      def isolation_level=(level)
        ···
        @scope =
          case level
          when :thread; Thread
          when :fiber; Fiber
          end
        ···
      end
      ...
      def context
        scope.current
      end
      ...
      self.isolation_level = :thread
    end
  end
end

# activerecord-8.0.5.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:713-715
def connection_lease
  @leases[ActiveSupport::IsolatedExecutionState.context]
end

# activerecord-8.0.5.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:409-428
def with_connection(prevent_permanent_checkout: false)
  lease = connection_lease # key 是 IsolatedExecutionState.context
  sticky_was = lease.sticky
  lease.sticky = false if prevent_permanent_checkout

  if lease.connection
    ...
  else
    begin
      yield lease.connection = checkout
    ensure
      lease.sticky = sticky_was if prevent_permanent_checkout && !sticky_was
      release_connection(lease) unless lease.sticky
    end
  end
end

# activerecord-8.0.5.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:866-887
def acquire_connection(checkout_timeout)
  if conn = @available.poll || try_to_checkout_new_connection   # 先拿空闲连接, 没有才尝试新建
    conn
  else
    reap                                                        # 回收死线程/纤程上的连接
    if conn = @available.poll || try_to_checkout_new_connection
      conn
    else
      @available.poll(checkout_timeout)                         # 还拿不到 → 排队等归还(下面 queue.rb)
    end
  end
end

# activerecord-8.0.5.1/lib/active_record/connection_adapters/abstract/connection_pool.rb:903-911
def try_to_checkout_new_connection
  ...
  if @threads_blocking_new_connections.zero? && (@connections.size + @now_connecting) < @size
    @now_connecting += 1    # 未达 @size(即 pool)才建新连接
  ...

# activerecord-8.0.5.1/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb:71-78
def poll(timeout = nil)
  synchronize { internal_poll(timeout) }
end

def internal_poll(timeout)
  no_wait_poll || (timeout && wait_poll(timeout))
end

# activerecord-8.0.5.1/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb:111-133
def wait_poll(timeout)
  ...
  if elapsed >= timeout
    msg = "could not obtain a connection from the pool within %0.3f seconds (waited %0.3f seconds); all pooled connections were in use" %
      [timeout, elapsed]
    raise ConnectionTimeoutError, msg
  end
  ...
end