Utilisation des planificateurs

Un planificateur contrôle le moment où un abonnement démarre et quand les notifications sont publiées. Il se compose de trois composants. Il s’agit d’abord d’une structure de données. Lorsque vous planifiez l’exécution des tâches, elles sont placées dans le planificateur pour la mise en file d’attente en fonction de la priorité ou d’autres critères. Il offre également un contexte d’exécution qui indique l’endroit où la tâche est exécutée (par exemple, dans le pool de threads, le thread actuel ou dans un autre domaine d’application). Enfin, il a une horloge qui fournit une notion de temps pour lui-même (en accédant à la Now propriété d’un planificateur). Les tâches planifiées sur un planificateur particulier respectent l’heure indiquée par cette horloge uniquement.

Les planificateurs introduisent également la notion de temps virtuel (désignée par le type VirtualScheduler), qui ne correspond pas au temps réel utilisé dans notre vie quotidienne. Par exemple, une séquence dont l’exécution est spécifiée pour prendre 100 ans peut être planifiée pour se terminer en temps virtuel en seulement 5 minutes. Cela sera abordé dans la rubrique Test et débogage des séquences observables .

Types de planificateurs

Les différents types de planificateurs fournis par Rx implémentent tous l’interface IScheduler . Chacun d’entre eux peut être créé et retourné à l’aide de propriétés statiques du type Scheduler. ImmediateScheduler (en accédant à la propriété Static Immediate) démarre immédiatement l’action spécifiée. Le CurrentThreadScheduler (en accédant à la propriété CurrentThread statique) planifie les actions à effectuer sur le thread qui effectue l’appel d’origine. L’action n’est pas exécutée immédiatement, mais elle est placée dans une file d’attente et exécutée uniquement une fois l’action actuelle terminée. DispatcherScheduler (en accédant à la propriété static Dispatcher) planifie les actions sur le répartiteur actuel, ce qui est bénéfique pour les développeurs Silverlight qui utilisent Rx. Les actions spécifiées sont ensuite déléguées à la méthode Dispatcher.BeginInvoke() dans Silverlight. NewThreadScheduler (en accédant à la propriété NewThread statique) planifie des actions sur un nouveau thread et est optimal pour la planification d’actions de longue durée ou de blocage. TaskPoolScheduler (en accédant à la propriété TaskPool statique) planifie des actions sur une fabrique de tâches spécifique. ThreadPoolScheduler (en accédant à la propriété ThreadPool statique) planifie des actions sur le pool de threads. Les deux planificateurs de pool sont optimisés pour les actions de courte durée.

Utilisation des planificateurs

Vous avez peut-être déjà utilisé des planificateurs dans votre code Rx sans indiquer explicitement le type de planificateurs à utiliser. Cela est dû au fait que tous les opérateurs Observable qui traitent l’accès concurrentiel ont plusieurs surcharges. Si vous n’utilisez pas la surcharge qui prend un planificateur comme argument, Rx choisit un planificateur par défaut en utilisant le principe de concurrence minimale. Cela signifie que le planificateur qui introduit le moins d’accès concurrentiel répondant aux besoins de l’opérateur est choisi.  Par exemple, pour les opérateurs retournant une observable avec un nombre limité et faible de messages, Rx appelle Immediate.  Pour les opérateurs qui retournent un nombre potentiellement élevé ou infini de messages, CurrentThread est appelé. Pour les opérateurs qui utilisent des minuteurs, ThreadPool est utilisé.

Étant donné que Rx utilise le planificateur le moins concurrentiel, vous pouvez choisir un autre planificateur si vous souhaitez introduire l’accès concurrentiel à des fins de performances, ou lorsque vous rencontrez un problème d’affinité de thread.  Un exemple de la première est que lorsque vous ne souhaitez pas bloquer un thread particulier, dans ce cas, vous devez utiliser ThreadPool.  Un exemple de ce dernier est que lorsque vous souhaitez qu’un minuteur s’exécute sur l’interface utilisateur, dans ce cas, vous devez utiliser Dispatcher. Pour spécifier un planificateur particulier, vous pouvez utiliser ces surcharges d’opérateur qui prennent un planificateur, par exemple, Timer(TimeSpan.FromSeconds(10), Scheduler.DispatcherScheduler()).

Dans l’exemple suivant, la séquence observable source produit des valeurs à un rythme effréné. La surcharge par défaut de l’opérateur timer place les messages OnNext sur le ThreadPool.

Observable.Timer(Timespan.FromSeconds(0.01))
          .Subscribe(…);

Cette opération sera rapidement mise en file d’attente sur l’observateur. Nous pouvons améliorer ce code à l’aide de l’opérateur ObserveOn, qui vous permet de spécifier le contexte que vous souhaitez utiliser pour envoyer des notifications push (OnNext) aux observateurs. Par défaut, l’opérateur ObserveOn garantit qu’OnNext sera appelé autant de fois que possible sur le thread actuel. Vous pouvez utiliser ses surcharges et rediriger les sorties OnNext vers un autre contexte. En outre, vous pouvez utiliser l’opérateur SubscribeOn pour retourner un proxy observable qui délègue des actions à un planificateur spécifique. Par exemple, pour une application gourmande en interface utilisateur, vous pouvez déléguer toutes les opérations en arrière-plan à effectuer sur un planificateur s’exécutant en arrière-plan en utilisant SubscribeOn et en lui transmettant un ThreadPoolScheduler. Pour recevoir des notifications envoyées et accéder à n’importe quel élément d’interface utilisateur, vous pouvez passer une instance de DispatcherScheduler à l’opérateur ObserveOn.

L’exemple suivant planifie toutes les notifications OnNext sur le répartiteur actuel, afin que toute valeur envoyée soit envoyée sur le thread d’interface utilisateur. Cela est particulièrement bénéfique pour les développeurs Silverlight qui utilisent Rx.

Observable.Timer(Timespan.FromSeconds(0.01))
          .ObserveOn(Scheduler.DispatcherScheduler)
          .Subscribe(…);

Au lieu d’utiliser l’opérateur ObserveOn pour modifier le contexte d’exécution sur lequel la séquence observable produit des messages, nous pouvons créer une concurrence au bon endroit pour commencer. À mesure que les opérateurs paramétrent l’introduction de l’accès concurrentiel en fournissant une surcharge d’argument du planificateur, la transmission du planificateur approprié entraîne moins d’endroits où l’opérateur ObserveOn doit être utilisé. Par exemple, nous pouvons débloquer l’observateur et nous abonner directement au thread d’interface utilisateur en modifiant le planificateur utilisé par la source, comme dans l’exemple suivant. Dans ce code, en utilisant la surcharge du minuteur qui prend un planificateur et en fournissant le Scheduler.Dispatcher instance, toutes les valeurs envoyées à partir de cette séquence observable proviennent du thread d’interface utilisateur.

Observable.Timer(Timespan.FromSeconds(0.01), Scheduler.DispatcherScheduler)
          .Subscribe(…);

Notez également qu’à l’aide de l’opérateur ObserveOn, une action est planifiée pour chaque message qui passe par la séquence observable d’origine. Cela peut modifier les informations de minutage et met davantage de pression sur le système. Si vous avez une requête qui compose différentes séquences observables s’exécutant sur de nombreux contextes d’exécution différents et que vous effectuez un filtrage dans la requête, il est préférable de placer ObserveOn ultérieurement dans la requête. En effet, une requête peut potentiellement filtrer un grand nombre de messages, et placer l’opérateur ObserveOn plus tôt dans la requête effectuerait un travail supplémentaire sur les messages qui seraient de toute façon filtrés. L’appel de l’opérateur ObserveOn à la fin de la requête aura un impact minimal sur les performances.

Un autre avantage de spécifier explicitement un type de planificateur est que vous pouvez introduire la concurrence à des fins de performances, comme illustré par le code suivant.

seq.GroupBy(...)
        .Select(x=>x.ObserveOn(Scheduler.NewThread))
        .Select(x=>expensive(x))  // perform operations that are expensive on resources