Rails源码研究之ActiveRecord:一，基本架构、CRUD封装与数据库连接

Rails的ORM框架ActiveRecord是马大叔的ActiveRecord模式的实现+associations+SingleTableInheritance
ActiveRecord的作者也是Rails的作者--David Heinemeier Hansson
ActiveRecord的key features:
1，零Meta Data，不需要XML配置文件
2，Database Support，现在支持mysql postgresql sqlite firebird sqlserver db2 oracle sybase openbase frontbase，写一个新的database adapter不会超过100行代码
3，线程安全，本地Ruby Web服务器，如WEBrick/Cerise，用线程处理请求
4，速度快，对100个对象循环查找一个值做benchmark，速度为直接数据库查询速度的50%
5，事务支持，使用事务来确保级联删除自动执行，同时也支持自己写事务安全的方法
6，简洁的关联，使用natural-language macros，如has_many、belongs_to
7，内建validations支持
8，自定义值对象

让我们深入研读一下ActiveRecord的核心源码
1，activerecord-1.15.3\lib\active_record.rb:

$:.unshift(File.dirname(__FILE__)) unless
  $:.include?(File.dirname(__FILE__)) || $:.include?(File.expand_path(File.dirname(__FILE__)))

unless defined?(ActiveSupport)
  begin
    $:.unshift(File.dirname(__FILE__) + "/../../activesupport/lib")  
    require 'active_support'  
  rescue LoadError
    require 'rubygems'
    gem 'activesupport'
  end
end

require 'active_record/base'
require 'active_record/observer'
require 'active_record/validations'
require 'active_record/callbacks'
require 'active_record/reflection'
require 'active_record/associations'
require 'active_record/aggregations'
require 'active_record/transactions'
require 'active_record/timestamp'
require 'active_record/acts/list'
require 'active_record/acts/tree'
require 'active_record/acts/nested_set'
require 'active_record/locking/optimistic'
require 'active_record/locking/pessimistic'
require 'active_record/migration'
require 'active_record/schema'
require 'active_record/calculations'
require 'active_record/xml_serialization'
require 'active_record/attribute_methods'

ActiveRecord::Base.class_eval do
  include ActiveRecord::Validations
  include ActiveRecord::Locking::Optimistic
  include ActiveRecord::Locking::Pessimistic
  include ActiveRecord::Callbacks
  include ActiveRecord::Observing
  include ActiveRecord::Timestamp
  include ActiveRecord::Associations
  include ActiveRecord::Aggregations
  include ActiveRecord::Transactions
  include ActiveRecord::Reflection
  include ActiveRecord::Acts::Tree
  include ActiveRecord::Acts::List
  include ActiveRecord::Acts::NestedSet
  include ActiveRecord::Calculations
  include ActiveRecord::XmlSerialization
  include ActiveRecord::AttributeMethods
end

unless defined?(RAILS_CONNECTION_ADAPTERS)
  RAILS_CONNECTION_ADAPTERS = %w( mysql postgresql sqlite firebird sqlserver db2 oracle sybase openbase frontbase )
end

RAILS_CONNECTION_ADAPTERS.each do |adapter|
  require "active_record/connection_adapters/" + adapter + "_adapter"
end

require 'active_record/query_cache'
require 'active_record/schema_dumper'

首先$:.unshift一句将当前文件加入动态库路径，然后确保加载ActiveSupport
然后将active_record/base/observer/validations.../attribute_methods等子目录下的文件require进来
然后用ActiveRecord::Base.class_eval将ActiveRecord::Validations/Locking/.../AttributeMethods等子模块include进来
RAILS_CONNECTION_ADAPTERS定义了ActiveRecord支持的database adapters的名字数组，然后循环将每个adapter文件require进来
最后将query_cache和schema_dumper这两个文件require进来

2，activerecord-1.15.3\lib\active_record\base.rb:

module ActiveRecord

  class Base

    class << self # Class methods

      def find(*args)
        options = extract_options_from_args!(args)
        validate_find_options(options)
        set_readonly_option!(options)

        case args.first
          when :first then find_initial(options)
          when :all   then find_every(options)
          else             find_from_ids(args, options)
        end
      end

      def find_by_sql(sql)
        connection.select_all(sanitize_sql(sql), "#{name} Load").collect! { |record| instantiate(record) }
      end

      def exists?(id_or_conditions)
        !find(:first, :conditions => expand_id_conditions(id_or_conditions)).nil?
      rescue ActiveRecord::ActiveRecordError
        false
      end

      def create(attributes = nil)
        if attributes.is_a?(Array)
          attributes.collect { |attr| create(attr) }
        else
          object = new(attributes)
          scope(:create).each { |att,value| object.send("#{att}=", value) } if scoped?(:create)
          object.save
          object
        end
      end

      def update(id, attributes)
        if id.is_a?(Array)
          idx = -1
          id.collect { |id| idx += 1; update(id, attributes[idx]) }
        else
          object = find(id)
          object.update_attributes(attributes)
          object
        end
      end

      def delete(id)
        delete_all([ "#{connection.quote_column_name(primary_key)} IN (?)", id ])
      end

      def destroy(id)
        id.is_a?(Array) ? id.each { |id| destroy(id) } : find(id).destroy
      end

      def update_all(updates, conditions = nil)
        sql  = "UPDATE #{table_name} SET #{sanitize_sql(updates)} "
        add_conditions!(sql, conditions, scope(:find))
        connection.update(sql, "#{name} Update")
      end

      def destroy_all(conditions = nil)
        find(:all, :conditions => conditions).each { |object| object.destroy }
      end

      def delete_all(conditions = nil)
        sql = "DELETE FROM #{table_name} "
        add_conditions!(sql, conditions, scope(:find))
        connection.delete(sql, "#{name} Delete all")
      end

      def count_by_sql(sql)
        sql = sanitize_conditions(sql)
        connection.select_value(sql, "#{name} Count").to_i
      end

      private

        def find_initial(options)
          options.update(:limit => 1) unless options[:include]
          find_every(options).first
        end

        def find_every(options)
          records = scoped?(:find, :include) || options[:include] ?
            find_with_associations(options) : 
            find_by_sql(construct_finder_sql(options))

          records.each { |record| record.readonly! } if options[:readonly]

          records
        end

        def find_from_ids(ids, options)
          expects_array = ids.first.kind_of?(Array)
          return ids.first if expects_array && ids.first.empty?

          ids = ids.flatten.compact.uniq

          case ids.size
            when 0
              raise RecordNotFound, "Couldn't find #{name} without an ID"
            when 1
              result = find_one(ids.first, options)
              expects_array ? [ result ] : result
            else
              find_some(ids, options)
          end
        end
      
        def find_one(id, options)
          conditions = " AND (#{sanitize_sql(options[:conditions])})" if options[:conditions]
          options.update :conditions => "#{table_name}.#{connection.quote_column_name(primary_key)} = #{quote_value(id,columns_hash[primary_key])}#{conditions}"

          if result = find_every(options).first
            result
          else
            raise RecordNotFound, "Couldn't find #{name} with ID=#{id}#{conditions}"
          end
        end
      
        def find_some(ids, options)
          conditions = " AND (#{sanitize_sql(options[:conditions])})" if options[:conditions]
          ids_list   = ids.map { |id| quote_value(id,columns_hash[primary_key]) }.join(',')
          options.update :conditions => "#{table_name}.#{connection.quote_column_name(primary_key)} IN (#{ids_list})#{conditions}"

          result = find_every(options)

          if result.size == ids.size
            result
          else
            raise RecordNotFound, "Couldn't find all #{name.pluralize} with IDs (#{ids_list})#{conditions}"
          end
        end

        def method_missing(method_id, *arguments)
          if match = /^find_(all_by|by)_([_a-zA-Z]\w*)$/.match(method_id.to_s)
            finder, deprecated_finder = determine_finder(match), determine_deprecated_finder(match)

            attribute_names = extract_attribute_names_from_match(match)
            super unless all_attributes_exists?(attribute_names)

            attributes = construct_attributes_from_arguments(attribute_names, arguments)

            case extra_options = arguments[attribute_names.size]
              when nil
                options = { :conditions => attributes }
                set_readonly_option!(options)
                ActiveSupport::Deprecation.silence { send(finder, options) }

              when Hash
                finder_options = extra_options.merge(:conditions => attributes)
                validate_find_options(finder_options)
                set_readonly_option!(finder_options)

                if extra_options[:conditions]
                  with_scope(:find => { :conditions => extra_options[:conditions] }) do
                    ActiveSupport::Deprecation.silence { send(finder, finder_options) }
                  end
                else
                  ActiveSupport::Deprecation.silence { send(finder, finder_options) }
                end

              else
                ActiveSupport::Deprecation.silence do
                  send(deprecated_finder, sanitize_sql(attributes), *arguments[attribute_names.length..-1])
                end
            end
          elsif match = /^find_or_(initialize|create)_by_([_a-zA-Z]\w*)$/.match(method_id.to_s)
            instantiator = determine_instantiator(match)
            attribute_names = extract_attribute_names_from_match(match)
            super unless all_attributes_exists?(attribute_names)

            attributes = construct_attributes_from_arguments(attribute_names, arguments)
            options = { :conditions => attributes }
            set_readonly_option!(options)

            find_initial(options) || send(instantiator, attributes)
          else
            super
          end
        end

        def extract_attribute_names_from_match(match)
          match.captures.last.split('_and_')
        end

        def construct_attributes_from_arguments(attribute_names, arguments)
          attributes = {}
          attribute_names.each_with_index { |name, idx| attributes[name] = arguments[idx] }
          attributes
        end

      protected

        def sanitize_sql(condition)
          case condition
            when Array; sanitize_sql_array(condition)
            when Hash;  sanitize_sql_hash(condition)
            else        condition
          end
        end

        def sanitize_sql_hash(attrs)
          conditions = attrs.map do |attr, value|
            "#{table_name}.#{connection.quote_column_name(attr)} #{attribute_condition(value)}"
          end.join(' AND ')

          replace_bind_variables(conditions, expand_range_bind_variables(attrs.values))
        end

        def sanitize_sql_array(ary)
          statement, *values = ary
          if values.first.is_a?(Hash) and statement =~ /:\w+/
            replace_named_bind_variables(statement, values.first)
          elsif statement.include?('?')
            replace_bind_variables(statement, values)
          else
            statement % values.collect { |value| connection.quote_string(value.to_s) }
          end
        end

        alias_method :sanitize_conditions, :sanitize_sql  

    end

    public

      def save
        create_or_update
      end

      def save!
        create_or_update || raise(RecordNotSaved)
      end

      def destroy
        unless new_record?
          connection.delete <<-end_sql, "#{self.class.name} Destroy"
            DELETE FROM #{self.class.table_name}
            WHERE #{connection.quote_column_name(self.class.primary_key)} = #{quoted_id}
          end_sql
        end

        freeze
      end

      def update_attribute(name, value)
        send(name.to_s + '=', value)
        save
      end

      def update_attributes(attributes)
        self.attributes = attributes
        save
      end

      def update_attributes!(attributes)
        self.attributes = attributes
        save!
      end

    private

      def create_or_update
        raise ReadOnlyRecord if readonly?
        result = new_record? ? create : update
        result != false
      end

      def update
        connection.update(
          "UPDATE #{self.class.table_name} " +
          "SET #{quoted_comma_pair_list(connection, attributes_with_quotes(false))} " +
          "WHERE #{connection.quote_column_name(self.class.primary_key)} = #{quote_value(id)}",
          "#{self.class.name} Update"
        )
      end

      def create
        if self.id.nil? && connection.prefetch_primary_key?(self.class.table_name)
          self.id = connection.next_sequence_value(self.class.sequence_name)
        end

        self.id = connection.insert(
          "INSERT INTO #{self.class.table_name} " +
          "(#{quoted_column_names.join(', ')}) " +
          "VALUES(#{attributes_with_quotes.values.join(', ')})",
          "#{self.class.name} Create",
          self.class.primary_key, self.id, self.class.sequence_name
        )

        @new_record = false
        id
      end

      def method_missing(method_id, *args, &block)
        method_name = method_id.to_s
        if @attributes.include?(method_name) or
            (md = /\?$/.match(method_name) and
            @attributes.include?(query_method_name = md.pre_match) and
            method_name = query_method_name)
          define_read_methods if self.class.read_methods.empty? && self.class.generate_read_methods
          md ? query_attribute(method_name) : read_attribute(method_name)
        elsif self.class.primary_key.to_s == method_name
          id
        elsif md = self.class.match_attribute_method?(method_name)
          attribute_name, method_type = md.pre_match, md.to_s
          if @attributes.include?(attribute_name)
            __send__("attribute#{method_type}", attribute_name, *args, &block)
          else
            super
          end
        else
          super
        end
      end

  end

end

base.rb这个文件比较大，它首先定义了Base类的Class Method，包括find、find_by_sql、create、update、destroy等
然后定义了一些private方法，如find_initial、find_every、find_from_ids等方法，它们供public的find方法调用
不出所料，private作用域里还定义了method_missing方法，它支持find_by_username、find_by_username_and_password、find_or_create_by_username等动态增加的方法
protected作用域里定义了sanitize_sql等辅助方法，这样子类(即我们的Model)中也可以使用这些protected方法
然后定义了Base类的public的Instance Method，如save、destroy、update_attribute、update_attributes等
然后定义了Base类的private的Instance Method，如供public的save方法调用的create_or_update、create、update等方法
然后定义了private的method_missing实例方法，供本类内其他实例方法访问本类的attributes

3，activerecord-1.15.3\lib\active_record\connection_adapters\abstract\connection_specification.rb:

module ActiveRecord
  class Base
    class ConnectionSpecification
      attr_reader :config, :adapter_method
      def initialize (config, adapter_method)
        @config, @adapter_method = config, adapter_method
      end
    end

    class << self

      def connection
        self.class.connection
      end

      def self.establish_connection(spec = nil)
      case spec
        when nil
          raise AdapterNotSpecified unless defined? RAILS_ENV
          establish_connection(RAILS_ENV)
        when ConnectionSpecification
          clear_active_connection_name
          @active_connection_name = name
          @@defined_connections[name] = spec
        when Symbol, String
          if configuration = configurations[spec.to_s]
            establish_connection(configuration)
          else
            raise AdapterNotSpecified, "#{spec} database is not configured"
          end
        else
          spec = spec.symbolize_keys
          unless spec.key?(:adapter) then raise AdapterNotSpecified, "database configuration does not specify adapter" end
          adapter_method = "#{spec[:adapter]}_connection"
          unless respond_to?(adapter_method) then raise AdapterNotFound, "database configuration specifies nonexistent #{spec[:adapter]} adapter" end
          remove_connection
          establish_connection(ConnectionSpecification.new(spec, adapter_method))
      end
    end
  end
end

connection_specification.rb文件定义了ActiveRecord::Base建立获取数据库连接相关的方法

4，activerecord-1.15.3\lib\active_record\connection_adapters\mysql_adapter.rb:

module ActiveRecord
  class Base
    def self.mysql_connection(config)
      config = config.symbolize_keys
      host = config[:host]
      port = config[:port]
      socket = config[:socket]
      username = config[:username]
      password = config[:password]

      if config.has_key?(:database)
        database = config[:database]
      else
        raise ArgumentError, "No database specified. Missing argument: database."
      end

      require_mysql
      mysql = Mysql.init
      mysql.ssl_set(config[:sslkey], config[:sslcert], config[:sslca], config[:sslcapath], config[:sslcipher]) if config[:sslkey]

      ConnectionAdapters::MysqlAdapter.new(mysql, logger, [host, username, password, database, port, socket], config)
    end
  end

  module ConnectionAdapters
    class MysqlAdapter < AbstractAdapter
      def initialize(connection, logger, connection_options, config)
        super(connection, logger)
        @connection_options, @config = connection_options, config

        connect
      end

      def execute(sql, name = nil) #:nodoc:
        log(sql, name) { @connection.query(sql) }
      rescue ActiveRecord::StatementInvalid => exception
        if exception.message.split(":").first =~ /Packets out of order/
          raise ActiveRecord::StatementInvalid, "'Packets out of order' error was received from the database. Please update your mysql bindings (gem install mysql) and read http://dev.mysql.com/doc/mysql/en/password-hashing.html for more information.  If you're on Windows, use the Instant Rails installer to get the updated mysql bindings."
        else
          raise
        end
      end

      def insert(sql, name = nil, pk = nil, id_value = nil, sequence_name = nil) #:nodoc:
        execute(sql, name = nil)
        id_value || @connection.insert_id
      end

      def update(sql, name = nil) #:nodoc:
        execute(sql, name)
        @connection.affected_rows
      end

      private
        def connect
          encoding = @config[:encoding]
          if encoding
            @connection.options(Mysql::SET_CHARSET_NAME, encoding) rescue nil
          end
          @connection.ssl_set(@config[:sslkey], @config[:sslcert], @config[:sslca], @config[:sslcapath], @config[:sslcipher]) if @config[:sslkey]
          @connection.real_connect(*@connection_options)
          execute("SET NAMES '#{encoding}'") if encoding

          execute("SET SQL_AUTO_IS_NULL=0")
        end
    end
  end
end

这个文件是mysql的数据库adapter的例子，其中mysql_connection->connect->real_connect方法会在establish_connection中调用

5，activerecord-1.15.3\lib\active_record\vendor\mysql.rb:

class Mysql

  def initialize(*args)
    @client_flag = 0
    @max_allowed_packet = MAX_ALLOWED_PACKET
    @query_with_result = true
    @status = :STATUS_READY
    if args[0] != :INIT then
      real_connect(*args)
    end
  end

  def real_connect(host=nil, user=nil, passwd=nil, db=nil, port=nil, socket=nil, flag=nil)
    @server_status = SERVER_STATUS_AUTOCOMMIT
    if (host == nil or host == "localhost") and defined? UNIXSocket then
      unix_socket = socket || ENV["MYSQL_UNIX_PORT"] || MYSQL_UNIX_ADDR
      sock = UNIXSocket::new(unix_socket)
      @host_info = Error::err(Error::CR_LOCALHOST_CONNECTION)
      @unix_socket = unix_socket
    else      
      sock = TCPSocket::new(host, port||ENV["MYSQL_TCP_PORT"]||(Socket::getservbyname("mysql","tcp") rescue MYSQL_PORT))
      @host_info = sprintf Error::err(Error::CR_TCP_CONNECTION), host
    end
    @host = host ? host.dup : nil
    sock.setsockopt Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, true
    @net = Net::new sock

    a = read
    @protocol_version = a.slice!(0)
    @server_version, a = a.split(/\0/,2)
    @thread_id, @scramble_buff = a.slice!(0,13).unpack("La8")
    if a.size >= 2 then
      @server_capabilities, = a.slice!(0,2).unpack("v")
    end
    if a.size >= 16 then
      @server_language, @server_status = a.slice!(0,3).unpack("cv")
    end

    flag = 0 if flag == nil
    flag |= @client_flag | CLIENT_CAPABILITIES
    flag |= CLIENT_CONNECT_WITH_DB if db

    @pre_411 = (0 == @server_capabilities & PROTO_AUTH41)
    if @pre_411
      data = Net::int2str(flag)+Net::int3str(@max_allowed_packet)+
             (user||"")+"\0"+
                   scramble(passwd, @scramble_buff, @protocol_version==9)
    else
      dummy, @salt2 = a.unpack("a13a12")
      @scramble_buff += @salt2
      flag |= PROTO_AUTH41
      data = Net::int4str(flag) + Net::int4str(@max_allowed_packet) +
             ([8] + Array.new(23, 0)).pack("c24") + (user||"")+"\0"+
             scramble41(passwd, @scramble_buff)
    end

    if db and @server_capabilities & CLIENT_CONNECT_WITH_DB != 0
      data << "\0" if @pre_411
      data << db
      @db = db.dup
    end
    write data
    pkt = read
    handle_auth_fallback(pkt, passwd)
    ObjectSpace.define_finalizer(self, Mysql.finalizer(@net))
    self
  end

  alias :connect :real_connect

  def real_query(query)
    command COM_QUERY, query, true
    read_query_result
    self
  end

  def query(query)
    real_query query
    if not @query_with_result then
      return self
    end
    if @field_count == 0 then
      return nil
    end
    store_result
  end

end

其中mysql.rb里的real_connect定义了Mysql数据库真正建立连接的方法

这次主要研究了ActiveRecord的基本架构、CRUD方法的封装以及以Mysql为例子的数据库连接相关的代码，歇会再聊，咳咳

评论

1 楼 fredzhang 2007-06-19  

不错，以前potian写了一系列route的源代码分析，可惜网站访问不了了，应当把这些源代码分析文章建立一个系列