¿Cómo se permite a NHibernate reintentar transacciones bloqueadas cuando se usa sesión por solicitud?

Question

¿Qué patrón/arquitectura usa en una aplicación de 3 niveles utilizando NHibernate que necesita admitir reintentos en fallas de transacción, cuando está utilizando el patrón de sesión por solicitud? (ya que ISession no es válida después de una excepción, incluso si se trata de un punto muerto o una excepción de tiempo de espera o de bloqueo de tiempo).

Answer 1

Nota: En la actualidad, nunca escribiría transacciones de escritura dentro del proyecto web, sino que utilizaría mensajería + colas y un trabajador en segundo plano que manejaría los mensajes con el objetivo de realizar el trabajo transaccional.

Sin embargo, todavía utilizaría transacciones para leer para obtener datos coherentes; junto con aislamiento MVCC/Snapshot, de proyectos web. En ese caso, encontrará que la sesión por solicitud por transacción está perfectamente bien.

Nota 1 Las ideas de esta publicación se han colocado en el Castle Transactions framework y mi nuevo NHibernate Facility.

OK, aquí está la idea general. Supongamos que desea crear un pedido no finalizado para un cliente. Tienes algún tipo de GUI, p. un navegador/MVC aplicación, que crean una nueva estructura de datos con la información relevante (o se obtiene esta estructura de datos de la red):

[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; } 
    } 
} 

se necesita algo para manejarlo. Probablemente este sería un controlador de comandos en un bus de servicio de algún tipo. La palabra 'controlador de comando' es una de muchas y también podría llamarlo 'servicio' o 'servicio de dominio' o 'manejador de mensajes'. Si estuvieras haciendo una programación funcional, sería tu implementación de cuadro de mensaje, o si estuvieras haciendo Erlang o Akka, sería un 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); 
} 

Lo anterior muestra un uso de la API puede elegir para este dominio determinado problema (manipulación de estado/transacción de la aplicación).

La implementación de Con:

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 
       } 
      } 
     } 
    } 
} 

Como se puede ver, necesitamos una nueva unidad de trabajo; la ISession cada vez que algo sale mal. Es por eso que el bucle está en el exterior de las instrucciones/bloques Using. Tener funciones equivale a tener instancias de fábrica, excepto que estamos invocando directamente en una instancia de objeto, en lugar de invocar un método sobre ella. Hace una mejor API de llamador.

Queremos un manejo bastante sencillo de la forma en que realizamos los reintentos, de modo que tenemos una interfaz que puede ser implementada por diferentes manejadores, llamada IRetryHandler. Debería ser posible encadenar estos para cada aspecto (sí, está muy cerca de AOP) que desea aplicar del flujo de control. De forma similar a como funciona el AOP, el valor de retorno se usa para controlar el flujo de control, pero solo de manera verdadera/falsa, que es nuestro requisito.

interface IRetryPolicy 
{ 
    bool PerformRetry(SqlException ex); 
} 

The AggregateRoot, PotentialCustomer es una entidad con una vida. Es lo que estaría mapeando con sus archivos * .hbm.xml/FluentNHibernate.

Tiene un método que corresponde a 1: 1 con el comando enviado. Esto hace que los controladores de comando sean completamente obvios para leer.

Además, con un lenguaje dinámico con tipado de pato, le permitiría asignar nombres de tipos de comandos a los métodos, de forma similar a como lo hace Ruby/Smalltalk.

Si estuvieras haciendo un evento, el manejo de la transacción sería similar, excepto que la transacción no interactuaría con NHibernate. El corolario es que guardarías los eventos creados mediante la invocación de CreateOrder (decimal) y le proporcionarías a tu entidad un mecanismo para volver a leer los eventos guardados de la tienda.

Un punto final a tener en cuenta es que estoy anulando tres métodos que he creado. Este es un requisito del lado de NHibernate, ya que necesita una forma de saber cuándo una entidad es igual a otra, en caso de que estén en conjuntos/bolsas. Más sobre mi implementación here. De cualquier manera, se trata de código de ejemplo y no me importa mi cliente en este momento, así que no me implementarlas:

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(); } 
} 

Necesitamos un método para crear políticas de reintento. Por supuesto que podríamos hacer esto de muchas maneras. Aquí estoy combinando una interfaz fluida con una instancia del mismo objeto del mismo tipo que el tipo de método estático. Implemento la interfaz explícitamente para que no haya otros métodos visibles en la interfaz fluida. Esta interfaz solo usa mis implementaciones de 'ejemplos' a continuación.

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}); 
    } 
} 

Necesitamos una interfaz para la invocación parcialmente completo a la interfaz fluida. Esto nos da seguridad de tipo. Por lo tanto, necesitamos dos operadores de desreferencia (es decir, 'punto final' - (.)), Lejos de nuestro tipo estático, antes de terminar de configurar la política.

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

La política de encadenamiento podría resolverse. Su implementación verifica que todos sus hijos regresen y, a medida que lo comprueba, también realiza la lógica en ellos.

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)); 
    } 
} 

Esta política permite que el hilo actual duerma un poco de tiempo; a veces la base de datos está sobrecargada, y tener múltiples lectores/escritores continuamente tratando de leer sería un ataque DOS de facto en la base de datos (vea lo que sucedió hace unos meses cuando Facebook se colgó porque todos sus servidores de caché consultaron sus bases de datos al mismo hora).

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; 
    } 
} 

De manera similar, en cualquier buen sistema basado en SQL, necesitamos manejar interbloqueos. Realmente no podemos planear esto en profundidad, especialmente cuando usamos NHibernate, aparte de mantener una política de transacción estricta, sin transacciones implícitas; y tenga cuidado con Open-Session-In-View. También está el problema del producto cartesiano/N + 1 selecciona el problema que necesitaría tener en cuenta si está buscando una gran cantidad de datos. En su lugar, puede tener Multi-Query o la palabra clave 'fetch' de HQL.

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; 
    } 
} 

Una clase de ayuda para que el código lea mejor.

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

No se olvide de controlar los errores de red en el IConnectionFactory así (delegando quizás a través de la implementación de IConnection).

---

PD: La sesión por solicitud es un patrón discontinuo si no solo está leyendo. Especialmente si está leyendo con la misma ISession con la que está escribiendo y no está ordenando las lecturas de modo que todas sean, siempre, antes de las escrituras.