ActiveRecord嵌套事务的实现

    class Transaction #:nodoc:
      attr_reader :connection, :state, :records, :savepoint_name, :isolation_level

      def initialize(connection, options, run_commit_callbacks: false)
       ...
        @joinable = options.fetch(:joinable, true)
       ...
      end

      def add_record(record)
        records << record
      end

    end

    class SavepointTransaction < Transaction
      def initialize(connection, savepoint_name, parent_transaction, *args)
        super(connection, *args)
        ...
        @savepoint_name = savepoint_name
      end

      def rollback
        # 回滚至标记点
        connection.rollback_to_savepoint(savepoint_name) if materialized?
        @state.rollback!
      end

      def commit
        # 释放还原点并提交
        connection.release_savepoint(savepoint_name) if materialized?
        @state.commit!
      end

    end

    class RealTransaction < Transaction
      def materialize!
        super
      end

      def rollback
        connection.rollback_db_transaction if materialized?
        @state.full_rollback!
      end

      def commit
        connection.commit_db_transaction if materialized?
        @state.full_commit!
      end
    end
    
    class TransactionManager #:nodoc:
      def initialize(connection)
        @stack = []
        @connection = connection
       ...
      end

      def begin_transaction(options = {})
        @connection.lock.synchronize do
          ...
          # 判断事务列表是否为空，如果不为空则表示存在嵌套事务，则使用SavepointTransactioin创建标记点
          transaction =
            if @stack.empty?
              RealTransaction.new(@connection, options, run_commit_callbacks: run_commit_callbacks)
            else
              SavepointTransaction.new(@connection, "active_record_#{@stack.size}", @stack.last, options,
                                       run_commit_callbacks: run_commit_callbacks)
            end

         ...
          @stack.push(transaction)
          transaction
        end
      end

      def commit_transaction
        @connection.lock.synchronize do
          transaction = @stack.last

          begin
            transaction.before_commit_records
          ensure
            @stack.pop
          end

          transaction.commit
          transaction.commit_records
        end
      end

      def rollback_transaction(transaction = nil)
        @connection.lock.synchronize do
          transaction ||= @stack.pop
          transaction.rollback
          transaction.rollback_records
        end
      end

      def within_new_transaction(options = {})
        @connection.lock.synchronize do
          transaction = begin_transaction options
          yield
        rescue Exception => error
          ...
        ensure
          ...
        end
      end

 def transaction(requires_new: nil, isolation: nil, joinable: true)
    # requires_new参数决定是否要创建嵌套事务，默认不创建嵌套事务，尝试将所有后续sql操作放到同一个事务中（即忽略多余的事务创建语句）
    if !requires_new && current_transaction.joinable?
      if isolation
        raise ActiveRecord::TransactionIsolationError, "cannot set isolation when joining a transaction"
      end
      yield
    else
      # 通过transaction_manager管理添加事务
      transaction_manager.within_new_transaction(isolation: isolation, joinable: joinable) { yield }
    end
  rescue ActiveRecord::Rollback
    # 捕获ActiveRecord::Rollback异常，使之不影响外层代码
    # rollbacks are silently swallowed
  end
end

User.transaction do
  User.create(:username => 'Kotori')
  # 不会创建新事务也不会创建savepoint
  User.transaction do
    # 这条操作会放到同一个事务中执行
    User.create(:username => 'Nemu')
    # ActiveRecord::Rollback被捕获，不会被外层代码块觉察，所以不会导致事务回滚，但是其它异常发生时仍会向上抛出，导致事务回滚
    raise ActiveRecord::Rollback
  end
end

User.transaction do
  User.create(:username => 'Kotori')
  # 创建savepoint模拟嵌套事务，不会创建真实的新事务(调用sql事务创建语句)
  User.transaction(:requires_new => true) do
    User.create(:username => 'Nemu')
    # ActiveRecord::Rollback在TransactionManager#within_new_transaction中被捕获并调用对应的rollback进行处理，回滚到标记点
    raise ActiveRecord::Rollback
  end
end

# Retrieve the connection associated with the current thread, or call
# #checkout to obtain one if necessary.
#
# #connection can be called any number of times; the connection is
# held in a cache keyed by a thread.
def connection
  @thread_cached_conns[connection_cache_key(@lock_thread || Thread.current)] ||= checkout
end