nhibernate - 使用 session 每个请求时，如何让 NHibernate 重试死锁事务？

Question

当您使用 Session-Per-Request 模式时，您在使用 NHibernate 的需要支持事务失败重试的 3 层应用程序中使用什么模式/架构？ (因为 ISession 在异常后变为无效，即使这是死锁、超时或活锁异常)。

Answer 1

注释 2 如今，我永远不会将写入事务放在 Web 项目中 - 而是使用消息传递 + 队列，并在后台处理消息的工作人员旨在完成事务性工作。

但是，我仍然会使用事务进行读取以获得一致的数据；与来自 Web 项目的 MVCC/Snapshot 隔离一起。在这种情况下，您会发现 session-per-request-per-transaction 非常好。

注释 1 这篇文章的想法已经放在Castle Transactions framework和我的新 NHibernate Facility .

好的，这是总体思路。假设您要为客户创建一个未完成的订单。你有某种图形用户界面，例如一个浏览器/MVC 应用程序，它使用相关信息创建一个新的数据结构(或者您从网络中获取此数据结构):

[Serializable]
class CreateOrder /*: IMessage*/
{
    // immutable
    private readonly string _CustomerName;
    private readonly decimal _Total;
    private readonly Guid _CustomerId;

    public CreateOrder(string customerName, decimal total, Guid customerId)
    {
        _CustomerName = customerName;
        _Total = total;
        _CustomerId = customerId;
    }

    // put ProtoBuf attribute
    public string CustomerName
    {
        get { return _CustomerName; }
    }

    // put ProtoBuf attribute
    public decimal Total
    {
        get { return _Total; }
    }

    // put ProtoBuf attribute
    public Guid CustomerId
    {
        get { return _CustomerId; }
    }
}

你需要一些东西来处理它。这可能是某种服务总线中的命令处理程序。 “命令处理程序”这个词是其中之一，您不妨将其称为“服务”或“域服务”或“消息处理程序”。如果你在做函数式编程，它会是你的消息框实现，或者如果你在做 Erlang 或 Akka，它会是一个 Actor。

class CreateOrderHandler : IHandle<CreateOrder>
{
    public void Handle(CreateOrder command)
    {
        With.Policy(IoC.Resolve<ISession>, s => s.BeginTransaction(), s =>
        {
            var potentialCustomer = s.Get<PotentialCustomer>(command.CustomerId);
            potentialCustomer.CreateOrder(command.Total);
            return potentialCustomer;
        }, RetryPolicies.ExponentialBackOff.RetryOnLivelockAndDeadlock(3));
    }
}

interface IHandle<T> /* where T : IMessage */
{
    void Handle(T command);
}

以上显示了您可能为此给定问题域(应用程序状态/事务处理)选择的 API 用法。

With的实现:

static class With
{
    internal static void Policy(Func<ISession> getSession,
                                       Func<ISession, ITransaction> getTransaction,
                                       Func<ISession, EntityBase /* abstract 'entity' base class */> executeAction,
                                       IRetryPolicy policy)
    {
        //http://fabiomaulo.blogspot.com/2009/06/improving-ado-exception-management-in.html

        while (true)
        {
            using (var session = getSession())
            using (var t = getTransaction(session))
            {
                var entity = executeAction(session);
                try
                {
                    // we might not always want to update; have another level of indirection if you wish
                    session.Update(entity);
                    t.Commit();
                    break; // we're done, stop looping
                }
                catch (ADOException e)
                {
                    // need to clear 2nd level cache, or we'll get 'entity associated with another ISession'-exception

                    // but the session is now broken in all other regards will will throw exceptions
                    // if you prod it in any other way
                    session.Evict(entity);

                    if (!t.WasRolledBack) t.Rollback(); // will back our transaction

                    // this would need to be through another level of indirection if you support more databases
                    var dbException = ADOExceptionHelper.ExtractDbException(e) as SqlException;

                    if (policy.PerformRetry(dbException)) continue;
                    throw; // otherwise, we stop by throwing the exception back up the layers
                }
            }
        }
    }
}

如您所见，我们需要一个新的工作单元； ISession 每次出现问题时。这就是循环位于 Using 语句/块之外的原因。拥有函数等同于拥有工厂实例，除了我们直接在对象实例上调用，而不是在其上调用方法。恕我直言，它提供了一个更好的调用者 API。

我们想要相当平滑地处理我们如何执行重试，因此我们有一个可以由不同处理程序实现的接口(interface)，称为 IRetryHandler。应该可以将这些链接到您想要强制执行控制流的每个方面(是的，它非常接近 AOP)。与 AOP 的工作方式类似，返回值用于控制控制流，但仅以真/假方式，这是我们的要求。

interface IRetryPolicy
{
    bool PerformRetry(SqlException ex);
}

AggregateRoot、PotentialCustomer 是一个具有生命周期的实体。这就是您将使用 *.hbm.xml 文件/FluentNHibernate 映射的内容。

它有一个方法与发送的命令 1:1 对应。这使得命令处理程序完全易于阅读。

此外，使用带有鸭子类型的动态语言，它允许您将命令的类型名称映射到方法，类似于 Ruby/Smalltalk 的做法。

如果你在做事件源，事务处理将是类似的，除了事务不会接口(interface) NHibernate 的。推论是，您将保存通过调用 CreateOrder(decimal) 创建的事件，并为您的实体提供一种从存储重新读取保存的事件的机制。

最后要注意的一点是我覆盖了我创建的三个方法。这是 NHibernate 方面的要求，因为它需要一种方法来知道一个实体何时等于另一个实体，如果它们在集合/包中。更多关于我的实现 here .无论如何，这是示例代码，我现在不关心我的客户，所以我没有实现它们:

sealed class PotentialCustomer : EntityBase
{
    public void CreateOrder(decimal total)
    {
        // validate total
        // run business rules

        // create event, save into event sourced queue as transient event
        // update private state
    }

    public override bool IsTransient() { throw new NotImplementedException(); }
    protected override int GetTransientHashCode() { throw new NotImplementedException(); }
    protected override int GetNonTransientHashCode() { throw new NotImplementedException(); }
}

我们需要一种创建重试策略的方法。当然，我们可以通过多种方式做到这一点。在这里，我将 fluent 接口(interface)与静态方法的类型相同类型的同一对象的实例相结合。我显式地实现了接口(interface)，以便在流畅的接口(interface)中看不到其他方法。此接口(interface)仅使用我下面的“示例”实现。

internal class RetryPolicies : INonConfiguredPolicy
{
    private readonly IRetryPolicy _Policy;

    private RetryPolicies(IRetryPolicy policy)
    {
        if (policy == null) throw new ArgumentNullException("policy");
        _Policy = policy;
    }

    public static readonly INonConfiguredPolicy ExponentialBackOff =
        new RetryPolicies(new ExponentialBackOffPolicy(TimeSpan.FromMilliseconds(200)));

    IRetryPolicy INonConfiguredPolicy.RetryOnLivelockAndDeadlock(int retries)
    {
        return new ChainingPolicy(new[] {new SqlServerRetryPolicy(retries), _Policy});
    }
}

我们需要一个接口(interface)来部分完成对流畅接口(interface)的调用。这给了我们类型安全。因此，在完成策略配置之前，我们需要两个取消引用运算符(即“句号”--(.))，远离我们的静态类型。

internal interface INonConfiguredPolicy
{
    IRetryPolicy RetryOnLivelockAndDeadlock(int retries);
}

链接策略可以解决。它的实现检查其所有子项是否返回 continue 并且在检查时，它还执行其中的逻辑。

internal class ChainingPolicy : IRetryPolicy
{
    private readonly IEnumerable<IRetryPolicy> _Policies;

    public ChainingPolicy(IEnumerable<IRetryPolicy> policies)
    {
        if (policies == null) throw new ArgumentNullException("policies");
        _Policies = policies;
    }

    public bool PerformRetry(SqlException ex)
    {
        return _Policies.Aggregate(true, (val, policy) => val && policy.PerformRetry(ex));
    }
}

这个策略让当前线程休眠一段时间；有时数据库过载，并且有多个读取器/写入器不断尝试读取将是对数据库的事实上的 DOS 攻击(看看几个月前 Facebook 崩溃时发生了什么，因为他们的缓存服务器都同时查询了他们的数据库时间)。

internal class ExponentialBackOffPolicy : IRetryPolicy
{
    private readonly TimeSpan _MaxWait;
    private TimeSpan _CurrentWait = TimeSpan.Zero; // initially, don't wait

    public ExponentialBackOffPolicy(TimeSpan maxWait)
    {
        _MaxWait = maxWait;
    }

    public bool PerformRetry(SqlException ex)
    {
        Thread.Sleep(_CurrentWait);
        _CurrentWait = _CurrentWait == TimeSpan.Zero ? TimeSpan.FromMilliseconds(20) : _CurrentWait + _CurrentWait;
        return _CurrentWait <= _MaxWait;
    }
}

同样，在任何良好的基于​​ SQL 的系统中，我们都需要处理死锁。我们不能真正深入地计划这些，尤其是在使用 NHibernate 时，除了保持严格的事务策略——没有隐式事务；并注意 Open-Session-In-View。如果您要获取大量数据，还需要牢记笛卡尔积问题/N+1 选择问题。取而代之的是，您可能有 Multi-Query 或 HQL 的 'fetch' 关键字。

internal class SqlServerRetryPolicy : IRetryPolicy
{
    private int _Tries;
    private readonly int _CutOffPoint;

    public SqlServerRetryPolicy(int cutOffPoint)
    {
        if (cutOffPoint < 1) throw new ArgumentOutOfRangeException("cutOffPoint");
        _CutOffPoint = cutOffPoint;
    }

    public bool PerformRetry(SqlException ex)
    {
        if (ex == null) throw new ArgumentNullException("ex");
        // checks the ErrorCode property on the SqlException
        return SqlServerExceptions.IsThisADeadlock(ex) && ++_Tries < _CutOffPoint;
    }
}

一个帮助类，使代码更好地阅读。

internal static class SqlServerExceptions
{
    public static bool IsThisADeadlock(SqlException realException)
    {
        return realException.ErrorCode == 1205;
    }
}

不要忘记处理 IConnectionFactory 中的网络故障(通过实现 IConnection 进行委托(delegate))。

PS:如果您不仅在阅读，则每次请求 session 是一种破坏模式。特别是如果您正在使用与您正在编写的同一个 ISession 进行阅读，并且您没有对读取进行排序，以便它们总是在写入之前。