Host del procesador de eventosEvent processor host

Azure Event Hubs es un eficaz servicio de ingesta de telemetría que se puede usar para hacer streaming de millones de eventos a un bajo costo.Azure Event Hubs is a powerful telemetry ingestion service that can be used to stream millions of events at low cost. En este artículo se describe cómo usar eventos ingeridos mediante el host del procesador de eventos (EPH), un agente de consumidor inteligente que simplifica la administración de la creación de puntos de comprobación, la concesión y los lectores de eventos paralelos.This article describes how to consume ingested events using the Event Processor Host (EPH); an intelligent consumer agent that simplifies the management of checkpointing, leasing, and parallel event readers.

La clave del escalado de Event Hubs es el concepto de consumidores con particiones.The key to scale for Event Hubs is the idea of partitioned consumers. En contraposición al patrón de consumidores de la competencia, el patrón de consumidores con particiones permite una alta escalabilidad mediante la eliminación de cuellos de botella de contención y la facilitación del paralelismo de principio a fin.In contrast to the competing consumers pattern, the partitioned consumer pattern enables high scale by removing the contention bottleneck and facilitating end to end parallelism.

Escenario de seguridad en el hogarHome security scenario

Como caso de ejemplo, considere una empresa de seguridad en el hogar que supervisa 100 000 casas.As an example scenario, consider a home security company that monitors 100,000 homes. Cada minuto obtiene los datos de los diversos sensores como el detector de movimiento, el sensor de apertura de puertas y ventanas, el detector de rotura de cristales, etc, instalados en cada casa.Every minute, it gets data from various sensors such as a motion detector, door/window open sensor, glass break detector, etc., installed in each home. La empresa proporciona un sitio web para que los residentes supervisen la actividad de su casa casi en tiempo real.The company provides a web site for residents to monitor the activity of their home in near real time.

Cada sensor inserta datos en un centro de eventos.Each sensor pushes data to an event hub. El centro de eventos está configurado con 16 particiones.The event hub is configured with 16 partitions. En el extremo de consumo, necesita un mecanismo que pueda leer estos eventos, consolidarlos (filtrarlos, agregarlos, etc.) y volcar el agregado a un blob de almacenamiento que, a continuación, se proyecta en una página web fácil de usar.On the consuming end, you need a mechanism that can read these events, consolidate them (filter, aggregate, etc.) and dump the aggregate to a storage blob, which is then projected to a user-friendly web page.

Escritura de la aplicación de consumidorWrite the consumer application

Al diseñar el consumidor en un entorno distribuido, el escenario debe controlar los siguientes requisitos:When designing the consumer in a distributed environment, the scenario must handle the following requirements:

  1. Escalado: cree varios consumidores y que cada consumidor tome posesión de la lectura desde varias particiones de Event Hubs.Scale: Create multiple consumers, with each consumer taking ownership of reading from a few Event Hubs partitions.
  2. Equilibrio de carga: aumente o reduzca dinámicamente los consumidores.Load balance: Increase or reduce the consumers dynamically. Por ejemplo, si se agrega un nuevo tipo de sensor (por ejemplo, un detector de monóxido de carbono) a cada casa, aumenta el número de eventos.For example, when a new sensor type (for example, a carbon monoxide detector) is added to each home, the number of events increases. En ese caso, el operador (una persona) aumenta el número de instancias de consumidor.In that case, the operator (a human) increases the number of consumer instances. A continuación, el grupo de consumidores puede volver a equilibrar el número de particiones que poseen para compartir la carga con los consumidores recién agregados.Then, the pool of consumers can rebalance the number of partitions they own, to share the load with the newly added consumers.
  3. Reanudación sin problemas después de los errores: si un consumidor (consumidor A) genera un error (por ejemplo, la máquina virtual que hospeda al consumidor, de repente, se bloquea), otros consumidores deben poder recopilar las particiones que posee el consumidor A y continuar.Seamless resume on failures: If a consumer (consumer A) fails (for example, the virtual machine hosting the consumer suddenly crashes), then other consumers must be able to pick up the partitions owned by consumer A and continue. Además, el punto de continuación, llamado punto de comprobación o de desplazamiento, debe estar en el punto exacto en el que se produjo el error del consumidor A o ligeramente antes.Also, the continuation point, called a checkpoint or offset, should be at the exact point at which consumer A failed, or slightly before that.
  4. Consumo de eventos: mientras que los tres puntos anteriores trataban sobre la administración del consumidor, también tiene que haber código para consumir los eventos y hacer algo útil con ellos como, por ejemplo, agregarlos y cargarlos en Blob Storage.Consume events: While the previous three points deal with the management of the consumer, there must be code to consume the events and do something useful with it; for example, aggregate it and upload it to blob storage.

En lugar de compilar su propia solución para esto, Event Hubs ofrece esta funcionalidad mediante la interfaz de IEventProcessor y la clase EventProcessorHost.Instead of building your own solution for this, Event Hubs provides this functionality through the IEventProcessor interface and the EventProcessorHost class.

Interfaz de IEventProcessorIEventProcessor interface

En primer lugar, las aplicaciones de consumo implementan la interfaz de IEventProcessor, que tiene cuatro métodos: OpenAsync, CloseAsync, ProcessErrorAsync y ProcessEventsAsync.First, consuming applications implement the IEventProcessor interface, which has four methods: OpenAsync, CloseAsync, ProcessErrorAsync, and ProcessEventsAsync. Esta interfaz contiene el código real para consumir los eventos que envía Event Hubs.This interface contains the actual code to consume the events that Event Hubs sends. El código siguiente muestra una implementación sencilla:The following code shows a simple implementation:

public class SimpleEventProcessor : IEventProcessor
{
    public Task CloseAsync(PartitionContext context, CloseReason reason)
    {
       Console.WriteLine($"Processor Shutting Down. Partition '{context.PartitionId}', Reason: '{reason}'.");
       return Task.CompletedTask;
    }

    public Task OpenAsync(PartitionContext context)
    {
       Console.WriteLine($"SimpleEventProcessor initialized. Partition: '{context.PartitionId}'");
       return Task.CompletedTask;
     }

    public Task ProcessErrorAsync(PartitionContext context, Exception error)
    {
       Console.WriteLine($"Error on Partition: {context.PartitionId}, Error: {error.Message}");
       return Task.CompletedTask;
    }

    public Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
    {
       foreach (var eventData in messages)
       {
          var data = Encoding.UTF8.GetString(eventData.Body.Array, eventData.Body.Offset, eventData.Body.Count);
             Console.WriteLine($"Message received. Partition: '{context.PartitionId}', Data: '{data}'");
       }
       return context.CheckpointAsync();
    }
}

A continuación, cree una instancia de una instancia de EventProcessorHost.Next, instantiate an EventProcessorHost instance. Según la sobrecarga, al crear la instancia de EventProcessorHost en el constructor, se usan los siguientes parámetros:Depending on the overload, when creating the EventProcessorHost instance in the constructor, the following parameters are used:

  • hostName: el nombre de cada instancia de consumidor.hostName: the name of each consumer instance. Cada instancia de EventProcessorHost debe tener un valor único para esta variable dentro de un grupo de consumidores, así que no codifique de forma rígida este valor.Each instance of EventProcessorHost must have a unique value for this variable within a consumer group, so don't hard code this value.
  • eventHubPath: El nombre del centro de eventos.eventHubPath: The name of the event hub.
  • consumerGroupName: Event Hubs usa $Default como nombre del grupo de consumidores predeterminado, pero es recomendable crear un grupo de consumidores para sus necesidades específicas de procesamiento.consumerGroupName: Event Hubs uses $Default as the name of the default consumer group, but it is a good practice to create a consumer group for your specific aspect of processing.
  • eventHubConnectionString: la cadena de conexión al centro de eventos que se puede recuperar desde Azure Portal.eventHubConnectionString: The connection string to the event hub, which can be retrieved from the Azure portal. Esta cadena de conexión debe tener permisos de escucha en el centro de eventos.This connection string should have Listen permissions on the event hub.
  • storageConnectionString: la cuenta de almacenamiento que se usa para la administración de recursos internos.storageConnectionString: The storage account used for internal resource management.

Por último, los consumidores registran la instancia de EventProcessorHost con el servicio Event Hubs.Finally, consumers register the EventProcessorHost instance with the Event Hubs service. Al registrar una clase de procesador de eventos con una instancia de EventProcessorHost, se inicia el procesamiento de eventos.Registering an event processor class with an instance of EventProcessorHost starts event processing. El proceso de registro indica al servicio Event Hubs que debe esperar que la aplicación del consumidor consuma eventos de algunas de sus particiones y que debe invocar el código de implementación de IEventProcessor siempre que envíe eventos para su consumo.Registering instructs the Event Hubs service to expect that the consumer app consumes events from some of its partitions, and to invoke the IEventProcessor implementation code whenever it pushes events to consume.

EjemploExample

Por ejemplo, imagine que hay 5 máquinas virtuales dedicadas a consumir eventos y una aplicación de consola simple en cada máquina virtual que es la que realiza el trabajo de consumo real.As an example, imagine that there are 5 virtual machines (VMs) dedicated to consuming events, and a simple console application in each VM, which does the actual consumption work. Cada aplicación de consola crea una instancia de EventProcessorHost y la registra con el servicio Event Hubs.Each console application then creates one EventProcessorHost instance and registers it with the Event Hubs service.

En este escenario de ejemplo, supongamos que se asignan 16 particiones a las 5 instancias de EventProcessorHost.In this example scenario, let's say that 16 partitions are allocated to the 5 EventProcessorHost instances. Algunas instancias de EventProcessorHost pueden disponer de algunas particiones más que otras.Some EventProcessorHost instances might own a few more partitions than others. Para cada partición que posee una instancia de EventProcessorHost esta crea una instancia de la clase SimpleEventProcessor.For each partition that an EventProcessorHost instance owns, it creates an instance of the SimpleEventProcessor class. Por tanto, hay 16 instancias de SimpleEventProcessor en total, con una de ellas asignada a cada partición.Therefore, there are 16 instances of SimpleEventProcessor overall, with one assigned to each partition.

En la siguiente lista se resume este ejemplo:The following list summarizes this example:

  • 16 particiones de Event Hubs.16 Event Hubs partitions.
  • 5 máquinas virtuales, 1 aplicación de consumidor (por ejemplo, Consumer.exe) en cada máquina virtual.5 VMs, 1 consumer app (for example, Consumer.exe) in each VM.
  • 5 instancias del host del procesador de eventos, 1 en cada máquina virtual con Consumer.exe.5 EPH instances registered, 1 in each VM by Consumer.exe.
  • 16 objetos SimpleEventProcessor que crean las 5 instancias del host del procesador de eventos.16 SimpleEventProcessor objects created by the 5 EPH instances.
  • 1 aplicación Consumer.exe puede contener 4 objetos SimpleEventProcessor, ya que 1 instancia del host del procesador de eventos puede poseer 4 particiones.1 Consumer.exe app might contain 4 SimpleEventProcessor objects, since the 1 EPH instance may own 4 partitions.

Seguimiento de la propiedad de una particiónPartition ownership tracking

Se puede realizar un seguimiento de la propiedad de una partición en una instancia del host del procesador de eventos (o un consumidor) mediante la cuenta de Azure Storage que se proporciona para tal fin.Ownership of a partition to an EPH instance (or a consumer) is tracked through the Azure Storage account that is provided for tracking. Puede visualizar el seguimiento en una sencilla tabla como se indica a continuación.You can visualize the tracking as a simple table, as follows. Puede ver la implementación real examinando los blobs en la cuenta de Storage proporcionada:You can see the actual implementation by examining the blobs under the Storage account provided:

Nombre del grupo de consumidoresConsumer group name Identificador de la particiónPartition ID Nombre de host (propietario)Host name (owner) Tiempo adquirido de concesión (o propiedad)Lease (or ownership) acquired time Desplazamiento en la partición (punto de comprobación)Offset in partition (checkpoint)
$Default$Default 00 Consumer_VM3Consumer_VM3 2018-04-15T01:23:452018-04-15T01:23:45 156156
$Default$Default 11 Consumer_VM4Consumer_VM4 2018-04-15T01:22:132018-04-15T01:22:13 734734
$Default$Default 22 Consumer_VM0Consumer_VM0 2018-04-15T01:22:562018-04-15T01:22:56 122122
::
::
$Default$Default 1515 Consumer_VM3Consumer_VM3 2018-04-15T01:22:562018-04-15T01:22:56 976976

En este caso, cada host adquiere la propiedad de una partición durante un determinado período de tiempo (la duración de la concesión).Here, each host acquires ownership of a partition for a certain duration (the lease duration). Si se produce un error en un host (se apaga la máquina virtual), la concesión expira.If a host fails (VM shuts down), then the lease expires. Otros hosts intentan obtener la propiedad de la partición y uno de ellos lo consigue.Other hosts try to get ownership of the partition, and one of the hosts succeeds. Este proceso restablece la concesión de la partición con un nuevo propietario.This process resets the lease on the partition with a new owner. De este modo, solo un lector a la vez puede leer en una determinada partición de un grupo de consumidores.This way, only a single reader at a time can read from any given partition within a consumer group.

Recepción de mensajesReceive messages

Cada llamada a ProcessEventsAsync ofrece una colección de eventos.Each call to ProcessEventsAsync delivers a collection of events. Es su responsabilidad administrar estos eventos.It is your responsibility to handle these events. Si desea asegurarse de que el host del procesador procesa cada mensaje al menos una vez, deberá escribir su propio código de reintento.If you want to make sure the processor host processes every message at least once, you need to write your own keep retrying code. Pero tenga cuidado con los mensajes dudosos.But be cautious about poisoned messages.

Se recomienda que el proceso se realice relativamente rápido, es decir, con el menor procesamiento posible.It is recommended that you do things relatively fast; that is, do as little processing as possible. En su lugar, utilice grupos de consumidores.Instead, use consumer groups. Si tiene que escribir en el almacenamiento y hacer algo de enrutamiento, es mejor usar dos grupos de consumidores y tener dos implementaciones de IEventProcessor que se ejecuten por separado.If you need to write to storage and do some routing, it is better to use two consumer groups and have two IEventProcessor implementations that run separately.

En algún momento durante el procesamiento, es posible que desee realizar un seguimiento de lo que ha leído y completado.At some point during your processing, you might want to keep track of what you have read and completed. Es fundamental realizar un seguimiento si debe reiniciar la lectura, para no tener que volver al principio de la transmisión.Keeping track is critical if you must restart reading, so you don't return to the beginning of the stream. EventProcessorHost simplifica este seguimiento mediante el uso de puntos de comprobación.EventProcessorHost simplifies this tracking by using checkpoints. Un punto de comprobación es una ubicación o desplazamiento, de una partición determinada, dentro de un grupo de consumidores determinado, en el que está satisfecho con los mensajes que se han procesado.A checkpoint is a location, or offset, for a given partition, within a given consumer group, at which point you are satisfied that you have processed the messages. Para marcar un punto de comprobación en EventProcessorHost llame al método CheckpointAsync en el objeto PartitionContext.Marking a checkpoint in EventProcessorHost is accomplished by calling the CheckpointAsync method on the PartitionContext object. Esta operación se realiza en el método ProcessEventsAsync pero también se puede realizar en CloseAsync.This operation is done within the ProcessEventsAsync method but can also be done in CloseAsync.

Puntos de controlCheckpointing

El método CheckpointAsync tiene dos sobrecargas: la primera, sin parámetros, crea un punto de comprobación en el desplazamiento del evento más alto de la colección que devuelve ProcessEventsAsync.The CheckpointAsync method has two overloads: the first, with no parameters, checkpoints to the highest event offset within the collection returned by ProcessEventsAsync. Este desplazamiento es una marca de "límite superior" que supone que ha procesado todos los eventos recientes cuando lo llama.This offset is a "high water" mark; it assumes you have processed all recent events when you call it. Si usa este método de esta manera, tenga en cuenta que se espera que lo llame una vez que ha devuelto el otro código de procesamiento de eventos.If you use this method in this way, be aware that you are expected to call it after your other event processing code has returned. La segunda sobrecarga le permite especificar una instancia de EventData para crear un punto de comprobación.The second overload lets you specify an EventData instance to checkpoint. Este método le permite usar un tipo diferente de marca de agua para crear un punto de comprobaciónThis method enables you to use a different type of watermark to checkpoint. Con esta marca de agua, puede implementar una marca "de límite inferior": el evento más bajo de la secuencia que está seguro de haber procesado.With this watermark, you can implement a "low water" mark: the lowest sequenced event you are certain has been processed. Esta sobrecarga se proporciona para ofrecer flexibilidad en la administración de desplazamientos.This overload is provided to enable flexibility in offset management.

Cuando se realiza la creación de un punto de comprobación, se escribe un archivo JSON con información específica de la partición en la cuenta de almacenamiento que se proporcionó en el constructor de EventProcessorHost.When the checkpoint is performed, a JSON file with partition-specific information (specifically, the offset), is written to the storage account supplied in the constructor to EventProcessorHost. Este archivo se actualiza continuamente.This file is continually updated. Es fundamental considerar la creación de puntos de comprobación en contexto. No es recomendable crear un punto de comprobación en cada mensaje.It is critical to consider checkpointing in context - it would be unwise to checkpoint every message. La cuenta de almacenamiento que se utilizó para la creación de puntos de comprobación no podrá administrar esta carga y, lo que es aún más importante, la creación de puntos de comprobación para todos los eventos puede indicar un patrón de mensajería en cola para el que una cola de Service Bus sería una opción mejor que un centro de eventos.The storage account used for checkpointing probably would not handle this load, but more importantly checkpointing every single event is indicative of a queued messaging pattern for which a Service Bus queue might be a better option than an event hub. La ventaja de Event Hubs es que obtiene al menos una entrega a gran escala.The idea behind Event Hubs is that you get "at least once" delivery at great scale. Al hacer los sistemas de nivel final idempotentes, es fácil recuperarse de errores o reinicios que hacen que los eventos se reciban múltiples veces.By making your downstream systems idempotent, it is easy to recover from failures or restarts that result in the same events being received multiple times.

Seguridad para subprocesos e instancias de procesadorThread safety and processor instances

De forma predeterminada, EventProcessorHost es un subproceso seguro y se comporta de una forma sincrónica con respecto a la instancia de IEventProcessor.By default, EventProcessorHost is thread safe and behaves in a synchronous manner with respect to the instance of IEventProcessor. Cuando llegan eventos para una partición, se llama a ProcessEventsAsync en la instancia de IEventProcessor de esa partición y se bloquean el resto de llamadas a ProcessEventsAsync de la partición.When events arrive for a partition, ProcessEventsAsync is called on the IEventProcessor instance for that partition and will block further calls to ProcessEventsAsync for the partition. Los siguientes mensajes y llamadas a ProcessEventsAsync se ponen en cola en segundo plano ya que el suministro de mensajes continúa ejecutándose en segundo plano en otros subprocesos.Subsequent messages and calls to ProcessEventsAsync queue up behind the scenes as the message pump continues to run in the background on other threads. Esta seguridad para subprocesos elimina la necesidad de colecciones seguras para subprocesos y aumenta considerablemente el rendimiento.This thread safety removes the need for thread-safe collections and dramatically increases performance.

Cierre correctoShut down gracefully

Por último, EventProcessorHost.UnregisterEventProcessorAsync permite un cierre correcto de todos los lectores de partición y se le debería llamar siempre al cerrar una instancia de EventProcessorHost.Finally, EventProcessorHost.UnregisterEventProcessorAsync enables a clean shutdown of all partition readers and should always be called when shutting down an instance of EventProcessorHost. Si no lo hace, puede provocar retrasos al iniciar otras instancias de EventProcessorHost debido a conflictos de época y de expiración de concesión.Failure to do so can cause delays when starting other instances of EventProcessorHost due to lease expiration and Epoch conflicts. La administración de épocas se trata detalladamente en la sección Época del artículo.Epoch management is covered in detail in the Epoch section of the article.

Administración de concesionesLease management

Al registrar una clase de procesador de eventos con una instancia de EventProcessorHost, se inicia el procesamiento de eventos.Registering an event processor class with an instance of EventProcessorHost starts event processing. La instancia de host obtiene las concesiones sobre algunas particiones del centro de eventos, posiblemente al tomar algunas de otras instancias, de una forma que converge en una distribución uniforme de particiones mediante todas las instancias de host.The host instance obtains leases on some partitions of the Event Hub, possibly grabbing some from other host instances, in a way that converges on an even distribution of partitions across all host instances. Para cada partición de la concesión, la instancia de host crea una instancia de la clase de procesador de eventos proporcionada, después, recibe eventos de esa partición y los pasa a la instancia de procesador de eventos.For each leased partition, the host instance creates an instance of the provided event processor class, then receives events from that partition, and passes them to the event processor instance. A medida que más casos se agregan y más concesiones se toman, EventProcessorHost equilibra finalmente la carga entre todos los consumidores.As more instances get added and more leases are grabbed, EventProcessorHost eventually balances the load among all consumers.

Como se explicó anteriormente, la tabla de seguimiento simplifica en gran medida la naturaleza de escalabilidad automática de EventProcessorHost.UnregisterEventProcessorAsync.As explained previously, the tracking table greatly simplifies the autoscale nature of EventProcessorHost.UnregisterEventProcessorAsync. Cuando se inicia una instancia de EventProcessorHost, esta adquiere el mayor número de concesiones posible y empieza a leer eventos.As an instance of EventProcessorHost starts, it acquires as many leases as possible, and begins reading events. A medida que la concesión se acerca a su fin, EventProcessorHost intenta renovarla realizando una reserva.As the leases near expiration, EventProcessorHost attempts to renew them by placing a reservation. Si la concesión está disponible para la renovación, el procesador continúa la lectura, pero si no es así, el lector se cierra y se realiza una llamada a CloseAsync.If the lease is available for renewal, the processor continues reading, but if it is not, the reader is closed and CloseAsync is called. CloseAsync es una buena oportunidad para realizar cualquier tarea de limpieza final para esa partición.CloseAsync is a good time to perform any final cleanup for that partition.

EventProcessorHost incluye una propiedad PartitionManagerOptions.EventProcessorHost includes a PartitionManagerOptions property. Esta propiedad habilita el control sobre la administración de concesiones.This property enables control over lease management. Establezca estas opciones antes de registrar la implementación de IEventProcessor.Set these options before registering your IEventProcessor implementation.

Control de las opciones del host del procesador de eventosControl Event Processor Host options

Además, una sobrecarga de RegisterEventProcessorAsync toma un objeto EventProcessorOptions como parámetro.Additionally, one overload of RegisterEventProcessorAsync takes an EventProcessorOptions object as a parameter. Utilice este parámetro para controlar el comportamiento del propio EventProcessorHost.UnregisterEventProcessorAsync.Use this parameter to control the behavior of EventProcessorHost.UnregisterEventProcessorAsync itself. EventProcessorOptions define cuatro propiedades y un evento:EventProcessorOptions defines four properties and one event:

  • MaxBatchSize: el tamaño máximo de la colección que desea recibir en una invocación de ProcessEventsAsync.MaxBatchSize: The maximum size of the collection you want to receive in an invocation of ProcessEventsAsync. Este tamaño no incluye el tamaño mínimo, solo el máximo.This size is not the minimum, only the maximum size. Si hay menos mensajes que recibir, ProcessEventsAsync se ejecuta con tantos como haya disponibles.If there are fewer messages to be received, ProcessEventsAsync executes with as many as were available.
  • PrefetchCount: un valor utilizado por el canal AMQP subyacente para determinar el límite superior de mensajes que debe recibir el cliente.PrefetchCount: A value used by the underlying AMQP channel to determine the upper limit of how many messages the client should receive. Este valor debe ser superior o igual a MaxBatchSize.This value should be greater than or equal to MaxBatchSize.
  • InvokeProcessorAfterReceiveTimeout: si este parámetro es true, se realizará una llamada a ProcessEventsAsync cuando se agote el tiempo de espera de la llamada subyacente para recibir eventos en una partición. Este método es útil para adoptar acciones basadas en tiempo durante los períodos de inactividad de la partición.InvokeProcessorAfterReceiveTimeout: If this parameter is true, ProcessEventsAsync is called when the underlying call to receive events on a partition times out. This method is useful for taking time-based actions during periods of inactivity on the partition.
  • InitialOffsetProvider: permite que se establezca un puntero de función o expresión lambda que se usa para proporcionar el desplazamiento inicial cuando un lector comienza la lectura de una partición.InitialOffsetProvider: Enables a function pointer or lambda expression to be set, which is called to provide the initial offset when a reader begins reading a partition. Si no se especifica este desplazamiento, el lector se iniciará en el evento más antiguo, a menos que ya se haya guardado un archivo JSON con un desplazamiento en la cuenta de almacenamiento que se proporcionó al constructor de EventProcessorHost.Without specifying this offset, the reader starts at the oldest event, unless a JSON file with an offset has already been saved in the storage account supplied to the EventProcessorHost constructor. Este método es útil cuando desea cambiar el comportamiento de inicio del lector.This method is useful when you want to change the behavior of the reader startup. Cuando se invoca este método, el parámetro de objeto contiene el identificador de partición para el que se inicia el lector.When this method is invoked, the object parameter contains the partition ID for which the reader is being started.
  • ExceptionReceivedEventArgs: le permite recibir notificaciones sobre cualquier excepción subyacente que se produzca en EventProcessorHost.ExceptionReceivedEventArgs: Enables you to receive notification of any underlying exceptions that occur in EventProcessorHost. Si las cosas no funcionan según lo esperado, este evento es un buen lugar para empezar a buscar.If things are not working as you expect, this event is a good place to start looking.

ÉpocaEpoch

Así es cómo funciona la época de recepción:Here is how the receive epoch works:

Con épocaWith Epoch

La época es un identificador único (valor de tiempo) que usa el servicio para aplicar la propiedad de la partición o de la concesión.Epoch is a unique identifier (epoch value) that the service uses, to enforce partition/lease ownership. Para crear un receptor de época se usa el método CreateEpochReceiver.You create an Epoch-based receiver using the CreateEpochReceiver method. Este método crea un receptor basado en época.This method creates an Epoch-based receiver. El receptor se crea para una partición de centro de eventos específica desde el grupo de consumidores especificado.The receiver is created for a specific event hub partition from the specified consumer group.

La característica de época ofrece a los usuarios la posibilidad de garantizar que solo hay un receptor en un grupo de consumidores en cualquier momento dado, con las siguientes reglas:The epoch feature provides users the ability to ensure that there is only one receiver on a consumer group at any point in time, with the following rules:

  • Si no hay ningún receptor existente en un grupo de consumidores, el usuario puede crear uno con cualquier valor de época.If there is no existing receiver on a consumer group, the user can create a receiver with any epoch value.
  • Si hay un receptor con un valor de época de e1 y se crea un receptor con un valor de época de e2 donde e1 < = e2, el receptor con e1 se desconectará automáticamente si el receptor con e2 se crea correctamente.If there is a receiver with an epoch value e1 and a new receiver is created with an epoch value e2 where e1 <= e2, the receiver with e1 will be disconnected automatically, receiver with e2 is created successfully.
  • Si hay un receptor con un valor de época de e1 y se crea un receptor con un valor de época de e2 donde e1 > e2 y, a continuación, la creación de e2 generará un error que indica que ya existe un receptor con la época e1.If there is a receiver with an epoch value e1 and a new receiver is created with an epoch value e2 where e1 > e2, then creation of e2 with fail with the error: A receiver with epoch e1 already exists.

No hay épocaNo Epoch

Creará un receptor no basado en época mediante el método CreateReceiver.You create a non-Epoch-based receiver using the CreateReceiver method.

Hay algunos escenarios en el procesamiento de streaming en los que a los usuarios les gustaría crear varios receptores en un solo grupo de consumidores.There are some scenarios in stream processing where users would like to create multiple receivers on a single consumer group. Para admitir estos escenarios, tenemos la posibilidad de crear un receptor sin época y, en este caso, se permiten hasta cinco receptores simultáneos en el grupo de consumidores.To support such scenarios, we do have ability to create a receiver without epoch and in this case we allow upto 5 concurrent receivers on the consumer group.

Modo mixtoMixed Mode

No recomendamos el uso de aplicaciones donde crea un receptor con época y, luego, cambia a sin época o viceversa, en el mismo grupo de consumidores.We don’t recommend application usage where you create a receiver with epoch and then switch to no-epoch or vice-versa on the same consumer group. Sin embargo, cuando se produce este comportamiento, el servicio lo controla mediante las reglas siguientes:However, when this behavior occurs, the service handles it using the following rules:

  • Si hay un receptor ya ha creado con época e1 y está recibiendo eventos activamente y se crea un receptor sin época, se producirá un error en la creación del receptor.If there is a receiver already created with epoch e1 and is actively receiving events and a new receiver is created with no epoch, the creation of new receiver will fail. Los receptores de época siempre tienen prioridad en el sistema.Epoch receivers always take precedence in the system.
  • Si hubiera un receptor ya ha creado con época e1 y se desconectara y se creara un receptor sin época en una nueva instancia de MessagingFactory, la creación del receptor se realizaría correctamente.If there was a receiver already created with epoch e1 and got disconnected, and a new receiver is created with no epoch on a new MessagingFactory, the creation of new receiver will succeed. Hay una salvedad aquí y es que nuestro sistema detectará la "desconexión del receptor" al cabo de unos 10 minutos.There is a caveat here that our system will detect the “receiver disconnection” after ~10 minutes.
  • Si hay uno o varios receptores creados sin época y se crea uno con época e1, se desconectan todos los receptores antiguos.If there are one or more receivers created with no epoch, and a new receiver is created with epoch e1, all the old receivers get disconnected.

Nota

Se recomienda el uso de diferentes grupos de consumidores para las aplicaciones que usan épocas y para aquellos que no usan épocas para evitar errores.We recommend using different consumer groups for applications that use epochs and for those that do not use epochs to avoid errors.

Pasos siguientesNext steps

Ahora que está familiarizado con el host del procesador de eventos, consulte los artículos siguientes para más información acerca de Event Hubs:Now that you're familiar with the Event Processor Host, see the following articles to learn more about Event Hubs: