Hibernate源码节选（三） Session的flush

关于ActionQueue

在hibernate中，Action可以理解为对数据库的操作。实体或集合的增、删、改都有对应的Action。ActionQueue会对这些Action进行分类保存。

public class ActionQueue { 
   
    private ExecutableList<AbstractEntityInsertAction> insertions;
    private ExecutableList<EntityDeleteAction> deletions;
    private ExecutableList<EntityUpdateAction> updates;

    private ExecutableList<CollectionRecreateAction> collectionCreations;
    private ExecutableList<CollectionUpdateAction> collectionUpdates;
    private ExecutableList<QueuedOperationCollectionAction> collectionQueuedOps;
    private ExecutableList<CollectionRemoveAction> collectionRemovals;

	.......
}

Session.flush

在了解ActionQueue之后，可以来看Session.flush了

public void flush() throws HibernateException { 
   
    checkOpen();//检查Ssession是否关闭，如已关闭则rollback
    doFlush();
}

熟悉的操作：生成event通知listener

private void doFlush() { 
   
    ......//一些事务的校验

    try { 
   
        ......//禁止在级联过程中flush

        FlushEvent flushEvent = new FlushEvent( this );
        for ( FlushEventListener listener : listeners( EventType.FLUSH ) ) { 
   
            listener.onFlush( flushEvent );//通知listener
        }
		......
    }
    catch ( RuntimeException e ) { 
   
        throw exceptionConverter.convert( e );
    }
}

• DefaultFlushEventListener#onFlush

监听器中主要是这三个方法

1.flushEverythingToExecutions 扫描实体和集合，生成Action，保存到ActionQueue

2.performExecutions 执行ActionQueue，生成SQL语句

3.postFlush 重新加载集合

public void onFlush(FlushEvent event) throws HibernateException { 
   
    final EventSource source = event.getSession();
    final PersistenceContext persistenceContext = source.getPersistenceContext();

    if ( persistenceContext.getNumberOfManagedEntities() > 0 ||
        persistenceContext.getCollectionEntries().size() > 0 ) { 
   

        try { 
   
            ......//flush前的一些预处理

            flushEverythingToExecutions( event );//1.生成Action，保存到ActinQueue
            performExecutions( source );//2.执行ActionQueue，生成SQL语句
            postFlush( source );//3.重新加载集合
        }
        finally { 
   
           ......
        }

        ......//flush后的一些处理
    }
}

1.生成Action，保存到ActionQueue

主要做了这四件事

1.prepareEntityFlushes 检查级联的对象，删除孤儿对象

2.prepareCollectionFlushes 检查待操作的集合是否有脏数据

3.flushEntities 找出脏实体、准备实体的UpdateAction、找出有效的集合

4.flushCollections 找出没有被引用的集合、扫描所有集合、准备集合的创建/删除/更新

protected void flushEverythingToExecutions(FlushEvent event) throws HibernateException { 
   
	
    ......
        
    EventSource session = event.getSession();

    final PersistenceContext persistenceContext = session.getPersistenceContext();
    
    ......

    prepareEntityFlushes( session, persistenceContext );
    // we could move this inside if we wanted to
    // tolerate collection initializations during
    // collection dirty checking:
    prepareCollectionFlushes( persistenceContext );
    // now, any collections that are initialized
    // inside this block do not get updated - they
    // are ignored until the next flush

    try { 
   
        int entityCount = flushEntities( event, persistenceContext );
        int collectionCount = flushCollections( session, persistenceContext );

        event.setNumberOfEntitiesProcessed( entityCount );
        event.setNumberOfCollectionsProcessed( collectionCount );
    }
    ......
}

1.1.prepareEntityFlushes 检查级联的对象，删除孤儿对象

比较复杂，看不太懂源码，占个坑先。

1.2.prepareCollectionFlushes 检查待操作的集合是否有脏数据

是否检查集合和集合元素是否可变，检查集合和是否含有脏数据

1.3.flushEntities 找出脏实体、准备实体的UpdateAction、找出有效的集合

为每个实体构建FlushEntityEvent事件，触发FlushEntityEventListener，创建EntityUpdateAction，放到队列中

对于实体里的集合属性，标记是否删除/创建/更新（但不放入Action队列中）、

1.4.flushCollections 找出没有被引用的集合、扫描所有集合、准备集合的创建/删除/更新

对所有集合做一次的标记

根据标记创建对应的Action，放入队列中

private int flushCollections(final EventSource session, final PersistenceContext persistenceContext) throws HibernateException { 
   
    ......

    final Map.Entry<PersistentCollection,CollectionEntry>[] entries = IdentityMap.concurrentEntries(
        (Map<PersistentCollection,CollectionEntry>) persistenceContext.getCollectionEntries()
    );

    final int count = entries.length;

    for ( Map.Entry<PersistentCollection,CollectionEntry> me : entries ) { 
   
        CollectionEntry ce = me.getValue();
        if ( !ce.isReached() && !ce.isIgnore() ) { 
   
            //检查集合并打上标记
            Collections.processUnreachableCollection( me.getKey(), session );
        }
    }

    // Schedule updates to collections:
	......

    ActionQueue actionQueue = session.getActionQueue();
    for ( Map.Entry<PersistentCollection,CollectionEntry> me :
         IdentityMap.concurrentEntries( (Map<PersistentCollection,CollectionEntry>) persistenceContext.getCollectionEntries() )) { 
   
        PersistentCollection coll = me.getKey();
        CollectionEntry ce = me.getValue();
		//根据标记创建对应的Action，放入队列中
        if ( ce.isDorecreate() ) { 
   
            session.getInterceptor().onCollectionRecreate( coll, ce.getCurrentKey() );
            actionQueue.addAction(
                new CollectionRecreateAction(
                    coll,
                    ce.getCurrentPersister(),
                    ce.getCurrentKey(),
                    session
                )
            );
        }
        /* 后面的依次为 * ce.isDoremove() 对应CollectionRemoveAction * ce.isDoupdate() 对应CollectionUpdateAction *( !coll.wasInitialized() && coll.hasQueuedOperations() ) 对应QueuedOperationCollectionAction */
        ......
    }
    actionQueue.sortCollectionActions();

    return count;
}

2.执行操作队列，生产SQL语句

以特定的顺序执行SQL语句（包括二级缓存的），这个顺序是为了保证不违反外键约束

protected void performExecutions(EventSource session) { 
   
   LOG.trace( "Executing flush" );

   // IMPL NOTE : here we alter the flushing flag of the persistence context to allow
   // during-flush callbacks more leniency in regards to initializing proxies and
   // lazy collections during their processing.
   // For more information, see HHH-2763
   try { 
   
      session.getJdbcCoordinator().flushBeginning();
      session.getPersistenceContext().setFlushing( true );
      // we need to lock the collection caches beforeQuery executing entity inserts/updates in order to
      // account for bi-directional associations
      session.getActionQueue().prepareActions();//给缓存集合加锁
      session.getActionQueue().executeActions();//按顺序执行队列
   }
   finally { 
   
      session.getPersistenceContext().setFlushing( false );
      session.getJdbcCoordinator().flushEnding();
   }
}

• 按顺序执行队列

1.INSERT

2.UPDATE

3.DELETE 集合

4.INSERT 集合

5.DELETE

利用LinkHashMap可以保持顺序的特性，依次取得各个操作队列的listProvider队头，然后遍历队内Action元素的execute方法，执行SQL语句。

public void executeActions() throws HibernateException { 
   
   if ( hasUnresolvedEntityInsertActions() ) { 
   
      throw new IllegalStateException( "About to execute actions, but there are unresolved entity insert actions." );
   }

   for ( ListProvider listProvider : EXECUTABLE_LISTS_MAP.values() ) { 
   
      ExecutableList<?> l = listProvider.get( this );
      if ( l != null && !l.isEmpty() ) { 
   
         executeActions( l );//最终会调用Action.execute()来执行SQL
      }
   }
}

EXECUTABLE_LISTS_MAP这个比较有意思，一个LinkHashMap，key使用Action类，value为ListProvider。可以理解为队头。这个map用来控制队列的执行顺序，从代码中可以看出更详细的队列分类及顺序。

private static final LinkedHashMap<Class<? extends Executable>,ListProvider> EXECUTABLE_LISTS_MAP;
static { 
   
    EXECUTABLE_LISTS_MAP = new LinkedHashMap<Class<? extends Executable>,ListProvider>( 8 );

    EXECUTABLE_LISTS_MAP.put(
        OrphanRemovalAction.class,
        new ListProvider<OrphanRemovalAction>() { 
   
            ExecutableList<OrphanRemovalAction> get(ActionQueue instance) { 
   
                return instance.orphanRemovals;
            }
            ExecutableList<OrphanRemovalAction> init(ActionQueue instance) { 
   
                // OrphanRemovalAction executables never require sorting.
                return instance.orphanRemovals = new ExecutableList<OrphanRemovalAction>( false );
            }
        }
    );
	}
	......
    /* * 剩下的依次为： * AbstractEntityInsertAction.class * EntityUpdateAction.class * QueuedOperationCollectionAction.class * CollectionRemoveAction.class * CollectionUpdateAction.class * CollectionRecreateAction.class * EntityDeleteAction.class */
}

突然发现Session.flush()到目前为止并没有提到EntityInsertAction的创建和进队？其实在Session.save()中已经创建EntityInsertAction并派到队伍中了。

3.重新加载集合

移除或者更新Session缓存中的集合

protected void postFlush(SessionImplementor session) throws HibernateException { 
   

    ......

    final PersistenceContext persistenceContext = session.getPersistenceContext();
    persistenceContext.getCollectionsByKey().clear();

    // the database has changed now, so the subselect results need to be invalidated
    // the batch fetching queues should also be cleared - especially the collection batch fetching one
    persistenceContext.getBatchFetchQueue().clear();

    for ( Map.Entry<PersistentCollection, CollectionEntry> me : IdentityMap.concurrentEntries( persistenceContext.getCollectionEntries() ) ) { 
   
        CollectionEntry collectionEntry = me.getValue();
        PersistentCollection persistentCollection = me.getKey();
        collectionEntry.postFlush(persistentCollection);
        if ( collectionEntry.getLoadedPersister() == null ) { 
   
            //if the collection is dereferenced, unset its session reference and remove from the session cache
            //iter.remove(); //does not work, since the entrySet is not backed by the set
            persistentCollection.unsetSession( session );
            persistenceContext.getCollectionEntries()
                .remove(persistentCollection);
        }
        else { 
   
            //otherwise recreate the mapping between the collection and its key
            CollectionKey collectionKey = new CollectionKey(
                collectionEntry.getLoadedPersister(),
                collectionEntry.getLoadedKey()
            );
            persistenceContext.getCollectionsByKey().put(collectionKey, persistentCollection);
        }
    }

现在SQL已经被写入到数据库中，剩下就等事务提交了。