聊聊hibernate的session-level repeatable reads

序

本文主要研究一下hibernate的session-level repeatable reads

實例

doInTransaction(session -> {
    Product product = new Product();
    product.setId(1L);
    product.setQuantity(7L);
    session.persist(product);
});
doInTransaction(session -> {
    final Product product = (Product) session.get(Product.class, 1L);
    try {
        executeSync(() -> doInTransaction(_session -> {
            Product otherThreadProduct = (Product) _session.get(Product.class, 1L);
            assertNotSame(product, otherThreadProduct);
            otherThreadProduct.setQuantity(6L);
        }));
        Product reloadedProduct = (Product) session.createQuery("from Product").uniqueResult();
        assertEquals(7L, reloadedProduct.getQuantity());
        assertEquals(6L, 
            ((Number) session
            .createSQLQuery("select quantity from product where id = :id")
            .setParameter("id", product.getId())
            .uniqueResult())
            .longValue()
        );
    } catch (Exception e) {
        fail(e.getMessage());
    }
});

• 這段代碼展現了hibernate的session-level repeatable reads功能，這裏reloadedProduct查詢返回的是session中id爲1的entity的緩存(可是也向db發出了sql語句，只是沒有使用其返回的resultSet的值)，而project操做查詢則直接根據jdbc查詢返回的結果返回

實例代碼來自 How does Hibernate guarantee application-level repeatable reads

write-behind

write-behind cache

• write-behind cache是cache策略的一種，其主要思路就是更新數據是首先更新cache，以後cache在批量持久化到存儲中，好比批量更新到數據，這樣作的好處是能夠合併數據的屢次操做減小IO

transactional write-behind cache

• hibernate爲了減小數據庫鏈接加鎖的時間，設計了transactional write-behind的策略，其persistence context充當transactional write-behind cache的角色，對entity的改動都先做用到內存，等到必定時機在flush到數據庫；具體體如今Session類中

To reduce lock contention in the database, the physical database transaction needs to be as short as possible.

Long-running database transactions prevent your application from scaling to a highly-concurrent load. Do not hold a database transaction open during end-user-level work, but open it after the end-user-level work is finished.
This concept is referred to as transactional write-behind.

The persistence context acts as a transactional write-behind cache, queuing any entity state change. Like any write-behind cache, changes are first applied in-memory and synchronized with the database during the flush time.

Session

hibernate的Session對jdbc的connection進行了包裝，它主要是維護了level one cache，即"repeatable read" persistence context；具體體如今Loader的getRow方法中

Behind the scenes, the Hibernate Session wraps a JDBC java.sql.Connection and acts as a factory for org.hibernate.Transaction instances. It maintains a generally "repeatable read" persistence context (first level cache) of the application domain model.

The Hibernate Session acts as a transaction-scoped cache providing repeatable reads for lookup by identifier and queries that result in loading entities.

Loader.getRow

hibernate-core-5.3.9.Final-sources.jar!/org/hibernate/loader/Loader.java

/**
     * Resolve any IDs for currently loaded objects, duplications within the
     * <tt>ResultSet</tt>, etc. Instantiate empty objects to be initialized from the
     * <tt>ResultSet</tt>. Return an array of objects (a row of results) and an
     * array of booleans (by side-effect) that determine whether the corresponding
     * object should be initialized.
     */
    private Object[] getRow(
            final ResultSet rs,
            final Loadable[] persisters,
            final EntityKey[] keys,
            final Object optionalObject,
            final EntityKey optionalObjectKey,
            final LockMode[] lockModes,
            final List hydratedObjects,
            final SharedSessionContractImplementor session) throws HibernateException, SQLException {
        final int cols = persisters.length;
        final EntityAliases[] descriptors = getEntityAliases();

        if ( LOG.isDebugEnabled() ) {
            LOG.debugf( "Result row: %s", StringHelper.toString( keys ) );
        }

        final Object[] rowResults = new Object[cols];

        for ( int i = 0; i < cols; i++ ) {

            Object object = null;
            EntityKey key = keys[i];

            if ( keys[i] == null ) {
                //do nothing
            }
            else {
                //If the object is already loaded, return the loaded one
                object = session.getEntityUsingInterceptor( key );
                if ( object != null ) {
                    //its already loaded so don't need to hydrate it
                    instanceAlreadyLoaded(
                            rs,
                            i,
                            persisters[i],
                            key,
                            object,
                            lockModes[i],
                            session
                    );
                }
                else {
                    object = instanceNotYetLoaded(
                            rs,
                            i,
                            persisters[i],
                            descriptors[i].getRowIdAlias(),
                            key,
                            lockModes[i],
                            optionalObjectKey,
                            optionalObject,
                            hydratedObjects,
                            session
                    );
                }
            }

            rowResults[i] = object;

        }

        return rowResults;
    }

• 在key不爲null的狀況下，該方法會設置object的值；這裏首先經過session.getEntityUsingInterceptor方法根據key從session中尋找該entity，若是不爲null，則執行instanceAlreadyLoaded，不然執行instanceNotYetLoaded去設置object

SessionImpl.getEntityUsingInterceptor

hibernate-core-5.3.9.Final-sources.jar!/org/hibernate/internal/SessionImpl.java

public final class SessionImpl
        extends AbstractSessionImpl
        implements EventSource, SessionImplementor, HibernateEntityManagerImplementor {

        //......

    @Override
    public Object getEntityUsingInterceptor(EntityKey key) throws HibernateException {
        checkOpenOrWaitingForAutoClose();
        // todo : should this get moved to PersistentContext?
        // logically, is PersistentContext the "thing" to which an interceptor gets attached?
        final Object result = persistenceContext.getEntity( key );
        if ( result == null ) {
            final Object newObject = getInterceptor().getEntity( key.getEntityName(), key.getIdentifier() );
            if ( newObject != null ) {
                lock( newObject, LockMode.NONE );
            }
            return newObject;
        }
        else {
            return result;
        }
    }

        //......
}

• getEntityUsingInterceptor方法首先從persistenceContext獲取entity，若是獲取不到再調用getInterceptor().getEntity獲取；若是沒有額外設置默認是EmptyInterceptor，其getEntity方法返回null

StatefulPersistenceContext

hibernate-core-5.3.9.Final-sources.jar!/org/hibernate/engine/internal/StatefulPersistenceContext.java

public class StatefulPersistenceContext implements PersistenceContext {
    //......

    // Loaded entity instances, by EntityKey
    private Map<EntityKey, Object> entitiesByKey;

    @Override
    public Object getEntity(EntityKey key) {
        return entitiesByKey.get( key );
    }

    @Override
    public void addEntity(EntityKey key, Object entity) {
        entitiesByKey.put( key, entity );
        if( batchFetchQueue != null ) {
            getBatchFetchQueue().removeBatchLoadableEntityKey(key);
        }
    }

    @Override
    public Object removeEntity(EntityKey key) {
        final Object entity = entitiesByKey.remove( key );
        final Iterator itr = entitiesByUniqueKey.values().iterator();
        while ( itr.hasNext() ) {
            if ( itr.next() == entity ) {
                itr.remove();
            }
        }
        // Clear all parent cache
        parentsByChild.clear();
        entitySnapshotsByKey.remove( key );
        nullifiableEntityKeys.remove( key );
        if( batchFetchQueue != null ) {
            getBatchFetchQueue().removeBatchLoadableEntityKey(key);
            getBatchFetchQueue().removeSubselect(key);
        }
        return entity;
    }

    @Override
    public void replaceDelayedEntityIdentityInsertKeys(EntityKey oldKey, Serializable generatedId) {
        final Object entity = entitiesByKey.remove( oldKey );
        final EntityEntry oldEntry = entityEntryContext.removeEntityEntry( entity );
        parentsByChild.clear();

        final EntityKey newKey = session.generateEntityKey( generatedId, oldEntry.getPersister() );
        addEntity( newKey, entity );
        addEntry(
                entity,
                oldEntry.getStatus(),
                oldEntry.getLoadedState(),
                oldEntry.getRowId(),
                generatedId,
                oldEntry.getVersion(),
                oldEntry.getLockMode(),
                oldEntry.isExistsInDatabase(),
                oldEntry.getPersister(),
                oldEntry.isBeingReplicated()
        );
    }

    //......
}

• StatefulPersistenceContext維護了一個entitiesByKey的map，getEntity方法直接根據EntityKey從該map取數據；它同時也提供了addEntity、removeEntity、replaceDelayedEntityIdentityInsertKeys等方法來修改map

instanceAlreadyLoaded

hibernate-core-5.3.9.Final-sources.jar!/org/hibernate/loader/Loader.java

public abstract class Loader {

    //......

    /**
     * The entity instance is already in the session cache
     */
    private void instanceAlreadyLoaded(
            final ResultSet rs,
            final int i,
            final Loadable persister,
            final EntityKey key,
            final Object object,
            final LockMode requestedLockMode,
            final SharedSessionContractImplementor session)
            throws HibernateException, SQLException {
        if ( !persister.isInstance( object ) ) {
            throw new WrongClassException(
                    "loaded object was of wrong class " + object.getClass(),
                    key.getIdentifier(),
                    persister.getEntityName()
            );
        }

        if ( LockMode.NONE != requestedLockMode && upgradeLocks() ) { //no point doing this if NONE was requested
            final EntityEntry entry = session.getPersistenceContext().getEntry( object );
            if ( entry.getLockMode().lessThan( requestedLockMode ) ) {
                //we only check the version when _upgrading_ lock modes
                if ( persister.isVersioned() ) {
                    checkVersion( i, persister, key.getIdentifier(), object, rs, session );
                }
                //we need to upgrade the lock mode to the mode requested
                entry.setLockMode( requestedLockMode );
            }
        }
    }

    //......
}

• instanceAlreadyLoaded方法主要是校驗類型是否正確，同時根據lockMode信息判斷是否要升級lock mode等

instanceNotYetLoaded

hibernate-core-5.3.9.Final-sources.jar!/org/hibernate/loader/Loader.java

public abstract class Loader {
    //......

    /**
     * The entity instance is not in the session cache
     */
    private Object instanceNotYetLoaded(
            final ResultSet rs,
            final int i,
            final Loadable persister,
            final String rowIdAlias,
            final EntityKey key,
            final LockMode lockMode,
            final EntityKey optionalObjectKey,
            final Object optionalObject,
            final List hydratedObjects,
            final SharedSessionContractImplementor session)
            throws HibernateException, SQLException {
        final String instanceClass = getInstanceClass(
                rs,
                i,
                persister,
                key.getIdentifier(),
                session
        );

        // see if the entity defines reference caching, and if so use the cached reference (if one).
        if ( session.getCacheMode().isGetEnabled() && persister.canUseReferenceCacheEntries() ) {
            final EntityDataAccess cache = persister.getCacheAccessStrategy();
            final Object ck = cache.generateCacheKey(
                    key.getIdentifier(),
                    persister,
                    session.getFactory(),
                    session.getTenantIdentifier()
                    );
            final Object cachedEntry = CacheHelper.fromSharedCache( session, ck, cache );
            if ( cachedEntry != null ) {
                CacheEntry entry = (CacheEntry) persister.getCacheEntryStructure().destructure( cachedEntry, factory );
                return ( (ReferenceCacheEntryImpl) entry ).getReference();
            }
        }

        final Object object;
        if ( optionalObjectKey != null && key.equals( optionalObjectKey ) ) {
            //its the given optional object
            object = optionalObject;
        }
        else {
            // instantiate a new instance
            object = session.instantiate( instanceClass, key.getIdentifier() );
        }

        //need to hydrate it.

        // grab its state from the ResultSet and keep it in the Session
        // (but don't yet initialize the object itself)
        // note that we acquire LockMode.READ even if it was not requested
        LockMode acquiredLockMode = lockMode == LockMode.NONE ? LockMode.READ : lockMode;
        loadFromResultSet(
                rs,
                i,
                object,
                instanceClass,
                key,
                rowIdAlias,
                acquiredLockMode,
                persister,
                session
        );

        //materialize associations (and initialize the object) later
        hydratedObjects.add( object );

        return object;
    }

    /**
     * Hydrate the state an object from the SQL <tt>ResultSet</tt>, into
     * an array or "hydrated" values (do not resolve associations yet),
     * and pass the hydrates state to the session.
     */
    private void loadFromResultSet(
            final ResultSet rs,
            final int i,
            final Object object,
            final String instanceEntityName,
            final EntityKey key,
            final String rowIdAlias,
            final LockMode lockMode,
            final Loadable rootPersister,
            final SharedSessionContractImplementor session) throws SQLException, HibernateException {

        final Serializable id = key.getIdentifier();

        // Get the persister for the _subclass_
        final Loadable persister = (Loadable) getFactory().getEntityPersister( instanceEntityName );

        if ( LOG.isTraceEnabled() ) {
            LOG.tracef(
                    "Initializing object from ResultSet: %s",
                    MessageHelper.infoString(
                            persister,
                            id,
                            getFactory()
                    )
            );
        }

        boolean fetchAllPropertiesRequested = isEagerPropertyFetchEnabled( i );

        // add temp entry so that the next step is circular-reference
        // safe - only needed because some types don't take proper
        // advantage of two-phase-load (esp. components)
        TwoPhaseLoad.addUninitializedEntity(
                key,
                object,
                persister,
                lockMode,
                session
        );

        //This is not very nice (and quite slow):
        final String[][] cols = persister == rootPersister ?
                getEntityAliases()[i].getSuffixedPropertyAliases() :
                getEntityAliases()[i].getSuffixedPropertyAliases( persister );

        final Object[] values = persister.hydrate(
                rs,
                id,
                object,
                rootPersister,
                cols,
                fetchAllPropertiesRequested,
                session
        );

        final Object rowId = persister.hasRowId() ? rs.getObject( rowIdAlias ) : null;

        final AssociationType[] ownerAssociationTypes = getOwnerAssociationTypes();
        if ( ownerAssociationTypes != null && ownerAssociationTypes[i] != null ) {
            String ukName = ownerAssociationTypes[i].getRHSUniqueKeyPropertyName();
            if ( ukName != null ) {
                final int index = ( (UniqueKeyLoadable) persister ).getPropertyIndex( ukName );
                final Type type = persister.getPropertyTypes()[index];

                // polymorphism not really handled completely correctly,
                // perhaps...well, actually its ok, assuming that the
                // entity name used in the lookup is the same as the
                // the one used here, which it will be

                EntityUniqueKey euk = new EntityUniqueKey(
                        rootPersister.getEntityName(), //polymorphism comment above
                        ukName,
                        type.semiResolve( values[index], session, object ),
                        type,
                        persister.getEntityMode(),
                        session.getFactory()
                );
                session.getPersistenceContext().addEntity( euk, object );
            }
        }

        TwoPhaseLoad.postHydrate(
                persister,
                id,
                values,
                rowId,
                object,
                lockMode,
                session
        );

    }

    //......
}

• instanceNotYetLoaded方法主要hydrate object，它會調用loadFromResultSet方法從resultSet提取出對象，而後添加到hydratedObjects中再返回；loadFromResultSet方法主要是經過TwoPhaseLoad.addUninitializedEntity方法，調用了session.getPersistenceContext().addEntity，將該object添加到StatefulPersistenceContext中；以後使用persister.hydrate從resultSet提取values，最後經過TwoPhaseLoad.postHydrate方法來建立managedEntity並與object關聯起來

initializeEntitiesAndCollections

hibernate-core-5.3.9.Final-sources.jar!/org/hibernate/loader/Loader.java

public abstract class Loader {
    //......

    private void initializeEntitiesAndCollections(
            final List hydratedObjects,
            final Object resultSetId,
            final SharedSessionContractImplementor session,
            final boolean readOnly,
            List<AfterLoadAction> afterLoadActions) throws HibernateException {

        final CollectionPersister[] collectionPersisters = getCollectionPersisters();
        if ( collectionPersisters != null ) {
            for ( CollectionPersister collectionPersister : collectionPersisters ) {
                if ( collectionPersister.isArray() ) {
                    //for arrays, we should end the collection load before resolving
                    //the entities, since the actual array instances are not instantiated
                    //during loading
                    //TODO: or we could do this polymorphically, and have two
                    //      different operations implemented differently for arrays
                    endCollectionLoad( resultSetId, session, collectionPersister );
                }
            }
        }

        //important: reuse the same event instances for performance!
        final PreLoadEvent pre;
        final PostLoadEvent post;
        if ( session.isEventSource() ) {
            pre = new PreLoadEvent( (EventSource) session );
            post = new PostLoadEvent( (EventSource) session );
        }
        else {
            pre = null;
            post = null;
        }

        if ( hydratedObjects != null ) {
            int hydratedObjectsSize = hydratedObjects.size();
            LOG.tracev( "Total objects hydrated: {0}", hydratedObjectsSize );
            for ( Object hydratedObject : hydratedObjects ) {
                TwoPhaseLoad.initializeEntity( hydratedObject, readOnly, session, pre );
            }
        }

        if ( collectionPersisters != null ) {
            for ( CollectionPersister collectionPersister : collectionPersisters ) {
                if ( !collectionPersister.isArray() ) {
                    //for sets, we should end the collection load after resolving
                    //the entities, since we might call hashCode() on the elements
                    //TODO: or we could do this polymorphically, and have two
                    //      different operations implemented differently for arrays
                    endCollectionLoad( resultSetId, session, collectionPersister );
                }
            }
        }

        // Until this entire method is refactored w/ polymorphism, postLoad was
        // split off from initializeEntity.  It *must* occur after
        // endCollectionLoad to ensure the collection is in the
        // persistence context.
        if ( hydratedObjects != null ) {
            for ( Object hydratedObject : hydratedObjects ) {
                TwoPhaseLoad.postLoad( hydratedObject, session, post );
                if ( afterLoadActions != null ) {
                    for ( AfterLoadAction afterLoadAction : afterLoadActions ) {
                        final EntityEntry entityEntry = session.getPersistenceContext().getEntry( hydratedObject );
                        if ( entityEntry == null ) {
                            // big problem
                            throw new HibernateException(
                                    "Could not locate EntityEntry immediately after two-phase load"
                            );
                        }
                        afterLoadAction.afterLoad( session, hydratedObject, (Loadable) entityEntry.getPersister() );
                    }
                }
            }
        }
    }

    //......
}

• initializeEntitiesAndCollections方法會調用TwoPhaseLoad.initializeEntity(該方法會調用persister.setPropertyValues( entity, hydratedState )來將hydratedState值填充到entity中)來初始化hydratedObject

小結

• write-behind cache是cache策略的一種，其主要思路就是更新數據是首先更新cache，以後cache在批量持久化到存儲中，好比批量更新到數據，這樣作的好處是能夠合併數據的屢次操做減小IO
• hibernate爲了減小數據庫鏈接加鎖的時間，設計了transactional write-behind的策略，其persistence context充當transactional write-behind cache的角色，對entity的改動都先做用到內存，等到必定時機在flush到數據庫；具體體如今Session類中
• hibernate的Session對jdbc的connection進行了包裝，它主要是維護了level one cache，即"repeatable read" persistence context；具體體如今Loader的getRow方法中
• 在key不爲null的狀況下，該方法會設置object的值；這裏首先經過session.getEntityUsingInterceptor方法根據key從session中尋找該entity，若是不爲null，則執行instanceAlreadyLoaded，不然執行instanceNotYetLoaded去設置object
• getEntityUsingInterceptor方法首先從persistenceContext獲取entity，若是獲取不到再調用getInterceptor().getEntity獲取；若是沒有額外設置默認是EmptyInterceptor，其getEntity方法返回null
• StatefulPersistenceContext維護了一個entitiesByKey的map，getEntity方法直接根據EntityKey從該map取數據；它同時也提供了addEntity、removeEntity、replaceDelayedEntityIdentityInsertKeys等方法來修改map
• instanceAlreadyLoaded方法主要是校驗類型是否正確，同時根據lockMode信息判斷是否要升級lock mode等；instanceNotYetLoaded方法主要hydrate object，它會調用loadFromResultSet方法從resultSet提取出對象，而後添加到hydratedObjects中再返回；loadFromResultSet方法主要是經過TwoPhaseLoad.addUninitializedEntity方法，調用了session.getPersistenceContext().addEntity，將該object添加到StatefulPersistenceContext中；以後使用persister.hydrate從resultSet提取values，最後經過TwoPhaseLoad.postHydrate方法來建立managedEntity並與object關聯起來
• initializeEntitiesAndCollections方法會調用TwoPhaseLoad.initializeEntity(該方法會調用persister.setPropertyValues( entity, hydratedState )來將hydratedState值填充到entity中)來初始化hydratedObject

doc

• 緩存更新的套路
• 極端事務處理模式：Write-behind 緩存
• Write-behind caching
• 6. Flushing
• 8.1. Physical Transactions
• A beginner’s guide to flush strategies in JPA and Hibernate
• How does Hibernate guarantee application-level repeatable reads