当您使用Session-Per-Request模式时,在使用NHibernate的3层应用程序中需要支持事务失败重试的情况下,您将使用什么模式/架构?(因为即使是死锁、超时或活锁异常,ISession在异常发生后也会变得无效)。
当您使用Session-Per-Request模式时,在使用NHibernate的3层应用程序中需要支持事务失败重试的情况下,您将使用什么模式/架构?(因为即使是死锁、超时或活锁异常,ISession在异常发生后也会变得无效)。
注意2 现在我永远不会将写事务放在Web项目中,而是使用消息传递+队列,并在后台使用工作程序处理消息,以实现事务性工作。
然而,对于从Web项目中获取一致数据的读取,我仍然会使用事务,结合MVCC/快照隔离。在这种情况下,您会发现每个请求的会话-每个事务都是完全可以的。
注意1 本文的思想已经被放置在Castle Transactions框架和我的新NHibernate设施中。
好的,这里是一般的想法。假设您想为客户创建一个未完成订单。您有某种GUI,例如浏览器/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; }
}
}
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更加友好。
我们希望能够平稳处理如何执行重试,因此我们有一个可以由不同处理程序实现的接口,称为IRetryHandler。对于要强制执行控制流程的每个方面(是的,它非常接近AOP),可以将这些链接在一起。类似于AOP的工作方式,返回值用于控制控制流,但仅以真/假方式进行,这是我们的要求。
interface IRetryPolicy
{
bool PerformRetry(SqlException ex);
}
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(); }
}
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});
}
}
internal interface INonConfiguredPolicy
{
IRetryPolicy RetryOnLivelockAndDeadlock(int retries);
}
链接策略可以被解决。其实现要检查所有子代是否返回继续,同时还执行其中的逻辑。
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));
}
}
此策略允许当前线程休眠一段时间;有时数据库会过载,如果多个读者/写者持续尝试读取将会对数据库造成拒绝服务攻击(请看几个月前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;
}
}
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进行委托处理)。
PS:如果你不仅仅是读取数据,那么每个请求一个Session的模式是有问题的。特别是当你使用相同的ISession进行读写操作时,并且没有按照顺序将读操作全部放在写操作之前。