Obserwatorzy

Czasami prosty wzorzec komunikatu/odpowiedzi nie wystarczy, a klient musi otrzymywać powiadomienia asynchroniczne. Na przykład użytkownik może chcieć otrzymywać powiadomienia, gdy znajomy publikuje nową wiadomość błyskawiczną.

Obserwatorzy klienta to mechanizm umożliwiający asynchroniczne powiadamianie klientów. Interfejsy obserwatora muszą dziedziczyć z IGrainObserver, a wszystkie metody muszą zwracać jedną z wartości: void, Task, Task<TResult>, ValueTask lub ValueTask<TResult>. Nie zalecamy zwracania typu void, ponieważ może to zachęcić do użycia async void w implementacji. Jest to niebezpieczny wzorzec, ponieważ może to spowodować awarię aplikacji, jeśli wyjątek zostanie rzucony w metodzie. Zamiast tego w przypadku scenariuszy powiadamiania o najlepszym wysiłku rozważ zastosowanie OneWayAttribute metody interfejsu obserwatora. Powoduje to, że odbiorca nie wysyła odpowiedzi na wywołanie metody i sprawia, że metoda natychmiast zwraca w miejscu wywołania, bez oczekiwania na odpowiedź ze strony obserwatora. Ziarno wywołuje metodę na obserwatorze, wywołując ją jak dowolną metodę interfejsu ziarna. Środowisko uruchomieniowe Orleans zapewnia prawidłowe dostarczanie żądań i odpowiedzi. Typowy przypadek użycia obserwatorów polega na zarejestrowaniu klienta w celu otrzymywania powiadomień w przypadku wystąpienia zdarzenia w Orleans aplikacji. Publikujący takie powiadomienia moduł powinien zapewnić interfejs API do dodawania lub usuwania obserwatorów. Ponadto zwykle wygodne jest uwidocznienie metody zezwalającej na anulowanie istniejącej subskrypcji.

Można użyć klasy narzędziowej, takiej jak ObserverManager<TObserver> , aby uprościć rozwój obserwowanych typów ziarna. W przeciwieństwie do ziaren, które Orleans automatycznie reaktywują się po awarii, klienci nie są odporni na błędy: klient, który zawiedzie, może nigdy nie odzyskać sprawności. Z tego powodu ObserverManager<T> narzędzie usuwa subskrypcje po skonfigurowanym czasie trwania. Aktywni klienci powinni odnawiać subskrypcje w określonych odstępach czasu, aby zachować ich aktywność.

Aby zasubskrybować powiadomienie, klient musi najpierw utworzyć obiekt lokalny implementowania interfejsu obserwatora. Następnie wywołuje metodę w fabryki obserwatora CreateObjectReference, aby przekształcić obiekt w referencję ziarna. Następnie możesz przekazać to odwołanie do metody subskrypcji na ziarnie powiadamiania.

Inne ziarna mogą również używać tego modelu do odbierania powiadomień asynchronicznych. Ziarna mogą implementować IGrainObserver interfejsy. W przeciwieństwie do przypadku subskrypcji klienta, ziarno subskrybujące implementuje interfejs obserwatora i przekazuje odwołanie do samego siebie (na przykład this.AsReference<IMyGrainObserverInterface>()). Nie ma potrzeby, że CreateObjectReference() ponieważ ziarna są już adresowalne.

Przykład kodu

Załóżmy, że masz ziarno, które okresowo wysyła komunikaty do klientów. Dla uproszczenia komunikat w naszym przykładzie jest ciągiem. Najpierw zdefiniuj interfejs na kliencie, który odbiera komunikat.

Interfejs wygląda następująco:

public interface IChat : IGrainObserver
{
    Task ReceiveMessage(string message);
}

Jedynym specjalnym wymaganiem jest to, że interfejs musi dziedziczyć z IGrainObserver. Teraz każdy klient, który chce obserwować te komunikaty, powinien zaimplementować klasę, która implementuje IChatelement .

Najprostszy przypadek wygląda następująco:

public class Chat : IChat
{
    public Task ReceiveMessage(string message)
    {
        Console.WriteLine(message);
        return Task.CompletedTask;
    }
}

Na serwerze należy mieć ziarno, które wysyła te wiadomości czatu do klientów. Ziarno powinno również zapewnić klientom mechanizm subskrybowania i anulowania subskrypcji powiadomień. W przypadku subskrypcji ziarno może używać wystąpienia klasy ObserverManager<TObserver> narzędziowej.

Uwaga

ObserverManager<TObserver> jest częścią Orleans począwszy od wersji 7.0. W przypadku starszych wersji można skopiować następującą implementację .

class HelloGrain : Grain, IHello
{
    private readonly ObserverManager<IChat> _subsManager;

    public HelloGrain(ILogger<HelloGrain> logger)
    {
        _subsManager =
            new ObserverManager<IChat>(
                TimeSpan.FromMinutes(5), logger);
    }

    // Clients call this to subscribe.
    public Task Subscribe(IChat observer)
    {
        _subsManager.Subscribe(observer, observer);

        return Task.CompletedTask;
    }

    //Clients use this to unsubscribe and no longer receive messages.
    public Task UnSubscribe(IChat observer)
    {
        _subsManager.Unsubscribe(observer);

        return Task.CompletedTask;
    }
}

Aby wysłać komunikat do klientów, użyj metody Notify wystąpienia ObserverManager<IChat>. Metoda przyjmuje metodę Action<T> lub wyrażenie lambda (gdzie T jest typu IChat tutaj). Możesz wywołać dowolną metodę w interfejsie, aby wysłać ją do klientów. W naszym przypadku mamy tylko jedną metodę , ReceiveMessagea nasz kod wysyłający na serwerze wygląda następująco:

public Task SendUpdateMessage(string message)
{
    _subsManager.Notify(s => s.ReceiveMessage(message));

    return Task.CompletedTask;
}

Teraz nasz serwer ma metodę wysyłania komunikatów do klientów obserwatorów i dwie metody subskrybowania/anulowania subskrypcji. Klient zaimplementował klasę umożliwiającą obserwowanie wiadomości Grain. Ostatnim krokiem jest utworzenie odwołania obserwatora na kliencie przy użyciu wcześniej zaimplementowanego Chat klasy i pozwolić na odbieranie komunikatów po zasubskrybowaniu.

Kod wygląda następująco:

//First create the grain reference
var friend = _grainFactory.GetGrain<IHello>(0);
Chat c = new Chat();

//Create a reference for chat, usable for subscribing to the observable grain.
var obj = _grainFactory.CreateObjectReference<IChat>(c);

//Subscribe the instance to receive messages.
await friend.Subscribe(obj);

Teraz za każdym razem, gdy nasz komponent na serwerze wywołuje metodę SendUpdateMessage, wszyscy subskrybenci otrzymują komunikat. W kodzie klienta, wystąpienie Chat w zmiennej c odbiera komunikat i wyświetla go w konsoli.

Ważne

Obiekty przekazywane do CreateObjectReference są przechowywane za pośrednictwem elementu WeakReference<T> i dlatego są bezużyteczne, jeśli nie istnieją żadne inne odwołania.

Należy zachować odniesienie dla każdego obserwatora, którego nie chcesz gromadzić.

Uwaga

Obserwatorzy są z natury zawodni, ponieważ klient hostujący obserwatora może zawieść, a obserwatorzy utworzeni po odbudowie mają różne (losowe) tożsamości. ObserverManager<TObserver> opiera się na okresowej ponownej subskrypcji przez obserwatorów, zgodnie z powyższym opisem, dzięki czemu może usunąć nieaktywnych obserwatorów.

Model wykonania

Implementacje IGrainObserver są rejestrowane za pośrednictwem wywołania IGrainFactory.CreateObjectReference. Każde wywołanie tej metody tworzy nowe odwołanie wskazujące tę implementację. Orleans wykonuje żądania wysyłane do każdego z tych odwołań jeden po jednym, aż do ukończenia. Obserwatorzy są nierentrancyjni; Orleans nie przeplata zatem równoczesnych żądań do obserwatora. Jeśli wielu obserwatorów odbiera żądania współbieżnie, te żądania mogą być wykonywane równolegle. Atrybuty, takie jak AlwaysInterleaveAttribute lub ReentrantAttribute nie wpływają na wykonywanie metod obserwatora; nie można dostosować modelu wykonywania.