java - Hibernate 卡在 tx.commit 上

Question

我有一个 hibernate 类，需要 3 个不同的 session 。它目前使用 2 个 session 并且运行良好。第一个 session 用于从外部数据库读取数据。第二个 session 用于将数据保存到我们的内部数据库。我添加第三个 session 是因为，无论主事务是否成功(XXXXUpdate 对象)，我们都需要跟踪事务。我的问题是新 session 卡在 tx.commit() 上。

private synchronized void executeUpdate(Long manualUpdateTagIndex) throws Exception {
    LogPersistenceLoggingContext ctx = new LogPersistenceThreadContext().getLogPersistenceLoggingContext();

    DateTime minTriggerDate = parseDateTimeIfNotNull(minTriggerTime);
    DateTime maxTriggerDate = parseDateTimeIfNotNull(maxTriggerTime);
    Session webdataSession = null;
    Session XXXXUpdateSession = null;
    XXXXUpdate update = new XXXXUpdate();
    update.setExecutedAt(new DateTime());
    update.setStatus(WebdataUpdateStatus.Success);

    boolean commit = true;
    int tagCount = 0;
    List<Period> tagPeriods = new ArrayList<>();
    Map<Long, DateTime> tagIndexes = new LinkedHashMap<>();

    try {

        XXXXUpdateSession = accountingService.openUnmanagedSession();
        XXXXUpdateSession.getTransaction().begin();
        XXXXUpdateSession.save(update);

        HierarchicalLogContext logCtx = new HierarchicalLogContext(String.valueOf(update.getId()));
        ctx.pushLoggingContext(logCtx);

        ctx.log(logger, Level.INFO, new XXXXLogMarker(), "Executing XXXX data transfer", new Object[]{});
        if (webdataSessionFactory == null){
            throw new Exception("Failed to obtain webdata session factory. See earlier log entries");
        }
        try {
            webdataSession = webdataSessionFactory.openSession();
        } catch (Exception ex) {
            update.setStatus(WebdataUpdateStatus.ConnectionError);
            throw new Exception("Failed to obtain webdata connection", ex);
        }

        webdataSession.getTransaction().begin();

        if (manualUpdateTagIndex == null) { // automatic tags update

            XXXXUpdate lastUpdate = (XXXXUpdate) HibernateUtil.getCurrentSpringManagedSession()
                    .createCriteria(XXXXUpdate.class)
                    .add(Restrictions.isNotNull("latestTriggerTimestamp"))
                    .add(Restrictions.eq("status", WebdataUpdateStatus.Success))
                    .add(Restrictions.eq("manualUpdate", false))
                    .addOrder(Order.desc("latestTriggerTimestamp"))
                    .setMaxResults(1).uniqueResult();

            DateTime lastUpdatedDate = Period.defaultEffectiveInstant;
            if (minTriggerDate != null) {
                lastUpdatedDate = minTriggerDate;
            }

            if (lastUpdate != null && lastUpdate.getLatestTriggerTimestamp() != null) {
                lastUpdatedDate = lastUpdate.getLatestTriggerTimestamp();
                ctx.log(logger, Level.INFO, new XXXXLogMarker(),
                        "Querying for tag event triggers newer than last update timestamp [" + lastUpdate.getLatestTriggerTimestamp() + "]", new Object[]{});
            } else {
                ctx.log(logger, Level.INFO, new XXXXLogMarker(), "Update has never run. Catching up with history", new Object[]{});
            }

            @SuppressWarnings("unchecked")
            List<XXXXProcessedTagRequest> processedReqs = HibernateUtil.getCurrentSpringManagedSession()
                    .createCriteria(XXXXProcessedTagRequest.class).list();

            Query triggerQuery = webdataSession.createQuery(
                    "select trigger, "
                            + "trigger.TagIndex,"
                            + "req  "
                            + "from XXXXTagEventTrigger as trigger "
                            + "join trigger.req as req "
                            + "where trigger.EventType in (:eventTypes) "
                            + "and trigger.timestamp > :lastUpdateMinusDelta "
                            + (maxTriggerDate != null?"and trigger.timestamp < :maxDate ":"")
                            + "and req.CurrentState = :currentState "
                            + "order by trigger.timestamp,trigger.reqIndex");

            triggerQuery.setParameterList("eventTypes", new Object[]{5, 9});
            triggerQuery.setParameter("lastUpdateMinusDelta", lastUpdatedDate.minusHours(hoursToKeepProcessedReqs) );
            if (maxTriggerDate != null){
                triggerQuery.setParameter("maxDate", maxTriggerDate);   
            }
            triggerQuery.setParameter("currentState", 2);

            @SuppressWarnings("unchecked")
            List<Object[]> allTriggers = triggerQuery.list();

            List<Object[]> unprocessedTriggers = removeProcessedTags(new ArrayList<Object[]>(allTriggers),processedReqs,ctx);

            for (Object[] row : unprocessedTriggers) {
                XXXXTagEventTrigger trigger = (XXXXTagEventTrigger) row[0];

                if (lastUpdatedDate == null || lastUpdatedDate.isBefore(trigger.getTimestamp().getMillis())) {
                    lastUpdatedDate = new DateTime(trigger.getTimestamp());
                }

                tagIndexes.put((Long) row[1], new DateTime(trigger.getTimestamp()));

                XXXXProcessedTagRequest processedReq = new XXXXProcessedTagRequest();
                processedReq.setReqIndex(((XXXXTagReq)row[2]).getReqIndex());
                processedReq.setTimestamp(trigger.getTimestamp());

                HibernateUtil.getCurrentSpringManagedSession().save(processedReq);
            }

            ctx.log(logger, Level.INFO, new XXXXLogMarker(),
                    "Found [" + unprocessedTriggers.size() + "] tag event triggers on [" + tagIndexes.size() + "] tags", new Object[]{});

            update.setLatestTriggerTimestamp(lastUpdatedDate);
        } else { // manual tag update
            ctx.log(logger, Level.INFO, new XXXXLogMarker(), "Executing manual update for tag index [" + manualUpdateTagIndex + "]", new Object[]{});

            DateTime now = new DateTime();
            tagIndexes.put(manualUpdateTagIndex, now);
            update.setLatestTriggerTimestamp(now);
            update.setManualUpdate(true);
        }

        if (tagIndexes.size() > 0) {

            int totalTagCount = tagIndexes.size();

            while (!tagIndexes.isEmpty()) {
                List<Long> batchIndexes = new ArrayList<>();
                Iterator<Map.Entry<Long, DateTime>> indexIt = tagIndexes.entrySet().iterator();

                while (indexIt.hasNext() && batchIndexes.size() < tagBatchSize) {
                    batchIndexes.add(indexIt.next().getKey());
                    indexIt.remove();
                }

                Map<Long, LocalTag> existingTags = new HashMap<>();
                @SuppressWarnings("unchecked")
                List<LocalTag> existingTagIds = HibernateUtil.getCurrentSpringManagedSession()
                        .createCriteria(LocalTag.class)
                        .add(Restrictions.in("tagIndex", batchIndexes))
                        .add(Restrictions.eq("currentVersion", true)).list();

                for (LocalTag lt : existingTagIds) {
                    existingTags.put(lt.getTagIndex(), lt);
                }

                ctx.log(logger, Level.INFO, new XXXXLogMarker(),
                        "Processing tag updates [" + tagCount + "-" + (tagCount + batchIndexes.size()) + "] of [" + totalTagCount + "]", new Object[]{});

                Criteria tagCriteria = webdataSession.createCriteria(XXXXTag.class);
                tagCriteria.add(Restrictions.in("TagIndex", batchIndexes));
                if (!includeTestTags) {
                    tagCriteria.add(Restrictions.eq("TestTag", "0"));
                }
                tagCriteria.setFetchMode("XXXXTagMS", FetchMode.JOIN);
                tagCriteria.setFetchMode("XXXXTagPS", FetchMode.JOIN);
                tagCriteria.setFetchMode("XXXXTagCCList", FetchMode.JOIN);
                tagCriteria.setFetchMode("XXXXTagTA", FetchMode.JOIN);
                tagCriteria.setFetchMode("XXXXTagCP", FetchMode.JOIN);
                tagCriteria.setResultTransformer(CriteriaSpecification.DISTINCT_ROOT_ENTITY);

                @SuppressWarnings("unchecked")
                List<XXXXTag> tags = tagCriteria.list();

                if (manualUpdateTagIndex != null && tags.isEmpty()) {
                    throw new ValidationException("No tag found for manual update tag index [" + manualUpdateTagIndex + "]");
                }

                for (XXXXTag tag : tags) {
                    update.getProcessedTags().add(updateTag(tag, tagIndexes.get(tag.getTagIndex()), existingTags));
                    tagCount++;
                    if (fireEventLastActions.contains(tag.getLastAction().trim())) {
                        tagPeriods.add(new Period(tag.getStartTime().getMillis(), tag.getStopTime().getMillis()));
                    }
                }

                HibernateUtil.getCurrentSpringManagedSession().flush();
                HibernateUtil.getCurrentSpringManagedSession().clear();

                webdataSession.clear();
            }
        } else {
            ctx.log(logger, Level.INFO, new XXXXLogMarker(), "No updates found", new Object[]{});
        }

        HibernateUtil.getCurrentSpringManagedSession()
        .createQuery("delete XXXXUpdate where executedAt < :purgeDate")
        .setParameter("purgeDate", new DateTime().minusDays(daysToKeepUpdateHistory))
        .executeUpdate();

        HibernateUtil.getCurrentSpringManagedSession()
        .createQuery("delete XXXXProcessedTagRequest where timestamp < :purgeDate")
        .setParameter("purgeDate", new DateTime().minusHours(hoursToKeepProcessedReqs))
        .executeUpdate();

        update.setStatus(WebdataUpdateStatus.Success);
        update.setTagCount(update.getProcessedTags().size());

        tagPeriods = Period.merge(tagPeriods);

        for (Period p : tagPeriods) {
            XXXXUpdatePeriod oup = new XXXXUpdatePeriod();
            oup.setXXXXUpdate(update);
            oup.setStartDate(p.getStart());
            oup.setEndDate(p.getEnd());
            update.getPeriods().add(oup);
        }

        HibernateUtil.getCurrentSpringManagedSession().flush();

        ctx.log(logger, Level.INFO, new XXXXLogMarker(), "XXXX data transfer complete. Transferred [" + tagCount + "] tag updates", new Object[]{});

        ctx.popLoggingContext(logCtx);
    } catch (Exception ex) {
        HibernateUtil.getCurrentSpringManagedSession().clear();
        update.getProcessedTags().clear();
        update.setTagCount(0);
        update.setStatus(WebdataUpdateStatus.TransferError);
        commit = false;
        ctx.log(logger, Level.ERROR, new XXXXLogMarker(), "XXXX data transfer failed", new Object[]{}, ex);
        throw new Exception("XXXX data transfer failed", ex);
    } finally {
        try {

            XXXXUpdateSession.saveOrUpdate(update);
            XXXXUpdateSession.getTransaction().commit();

        } catch (Exception ex) {
            commit = false;
            ctx.log(logger, Level.ERROR, new XXXXLogMarker(), "Failed to save XXXX transfer update record", new Object[]{}, ex);
            throw new Exception("Failed to save XXXX transfer update record", ex);
        } finally {
            if (!commit) {
                webdataSession.getTransaction().rollback();
            } else {
                webdataSession.getTransaction().commit();
            }
            ResourceDisposer.dispose(webdataSession);
        }

    }

}

新 session 是 XXXXUpdateSession。唯一的新代码是与本次 session 相关的代码。这是某种计时问题，因为当我使用 hibernate 调试日志记录时，tx 提交没有问题。当我尝试调试 hibernate commit() 时它也会提交。我对 hibernate 没有太多经验，所以我可能错过了一些明显的东西。任何帮助将不胜感激。谢谢。

Answer 1

您打开了两个事务 webdataSession.getTransaction().begin();，这导致了问题(上述代码中的 20 行和 37 行)。

您可以在提交第一个事务后打开第二个事务。

此外，使用冗长的方法也不是最佳实践，因为这样会很难调试问题并成为项目维护/支持的噩梦。