관찰자

  • 아티클

간단한 메시지/응답 패턴으로는 충분하지 않고 클라이언트가 비동기 알림을 받아야 하는 경우가 있습니다. 예를 들어 사용자는 친구가 새 인스턴트 메시지를 게시할 때 알림을 받고 싶을 수 있습니다.

클라이언트 관찰자는 클라이언트에 비동기적으로 알릴 수 있는 메커니즘입니다. 관찰자 인터페이스는 IGrainObserver에서 상속해야 하며 모든 메서드는 void, Task, Task<TResult>, ValueTask 또는 ValueTask<TResult>를 반환해야 합니다. void의 반환 형식은 구현에서 async void를 사용하도록 권장할 수 있으므로 권장되지 않습니다. 이는 메서드에서 예외가 throw될 경우 애플리케이션 크래시가 발생할 수 있으므로 위험한 패턴입니다. 대신 최상의 알림 시나리오의 경우 관찰자의 인터페이스 메서드에 OneWayAttribute를 적용하는 것이 좋습니다. 이렇게 하면 수신자가 메서드 호출에 대한 응답을 보내지 않으며 관찰자의 응답을 기다리지 않고 메서드가 호출 사이트에서 즉시 반환됩니다. grain은 모든 grain 인터페이스 메서드처럼 호출하여 관찰자에서 메서드를 호출합니다. Orleans 런타임은 요청 및 응답의 전달을 보장합니다. 관찰자의 일반적인 사용 사례는 Orleans 애플리케이션에서 이벤트가 발생할 때 알림을 수신하도록 클라이언트를 등록하는 것입니다. 이러한 알림을 게시하는 조직은 관찰자를 추가하거나 제거하는 API를 제공해야 합니다. 또한 일반적으로 기존 구독을 취소할 수 있는 메서드를 노출하는 것이 편리합니다.

조직 개발자는 관찰된 조직 형식의 개발을 간소화하는 ObserverManager<TObserver>와 같은 유틸리티 클래스를 사용할 수 있습니다. 오류 발생 후 필요에 따라 자동으로 다시 활성화되는 조직과 달리 클라이언트는 내결함성이 없으므로 실패한 클라이언트는 복구되지 않을 수 있습니다. 이러한 이유로 ObserverManager<T> 유틸리티는 구성된 기간 후에 구독을 제거합니다. 활성 상태인 클라이언트는 타이머를 다시 구독하여 구독을 활성 상태로 유지해야 합니다.

알림을 구독하려면 먼저 클라이언트가 관찰자 인터페이스를 구현하는 로컬 개체를 만들어야 합니다. 그런 다음, 관찰자 팩터리인 CreateObjectReference에서 메서드를 호출하여 개체를 조직 참조로 변환한 다음, 알림 조직의 구독 메서드에 전달할 수 있습니다.

이 모델은 다른 조직에서 비동기 알림을 받는 데 사용할 수도 있습니다. 조직은 IGrainObserver 인터페이스도 구현할 수도 있습니다. 클라이언트 구독의 경우와 달리 조직 구독은 단순히 관찰자 인터페이스를 구현하고 자체에 대한 참조를 전달합니다(예: this.AsReference<IMyGrainObserverInterface>()). 조직은 이미 주소 지정이 가능하기 때문에 CreateObjectReference()가 필요하지 않습니다.

코드 예제

주기적으로 클라이언트에 메시지를 보내는 조직이 있다고 가정해 보겠습니다. 간단한 설명을 위해 이 예제의 메시지는 문자열입니다. 먼저 클라이언트에서 메시지를 받을 인터페이스를 정의합니다.

인터페이스는 다음과 같습니다.

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

유일한 특별한 점은 인터페이스가 IGrainObserver에서 상속되어야 한다는 것입니다. 이제 해당 메시지를 관찰하려는 클라이언트는 IChat을 구현하는 클래스를 구현해야 합니다.

가능 간단한 경우는 다음과 같이 나타납니다.

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

서버에는 다음으로 이러한 채팅 메시지를 클라이언트로 보내는 조직이 있어야 합니다. 또한 조직에는 클라이언트가 알림을 구독하고 구독을 취소하는 메커니즘이 있어야 합니다. 구독의 경우 조직은 유틸리티 클래스 ObserverManager<TObserver>의 인스턴스를 사용할 수 있습니다.

참고 항목

ObserverManager<TObserver>는 버전 7.0 이후부터 Orleans에 포함되었습니다. 이전 버전의 경우 다음 구현을 복사할 수 있습니다.

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

클라이언트에 메시지를 보내기 위해 ObserverManager<IChat> 인스턴스의 Notify 메서드를 사용할 수 있습니다. 이 메서드는 Action<T> 메서드 또는 람다 식을 사용합니다(여기서 T는 IChat 형식임). 인터페이스에서 메서드를 호출하여 클라이언트에 보낼 수 있습니다. 이 경우 메서드(ReceiveMessage)가 하나뿐이며 서버에서 보내는 코드는 다음과 같습니다.

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

    return Task.CompletedTask;
}

이제 서버에는 관찰자 클라이언트에 메시지를 보내는 메서드와 구독/구독 취소를 위한 두 개의 메서드가 있으며 클라이언트는 조직 메시지를 관찰할 수 있는 클래스를 구현했습니다. 마지막 단계는 이전에 구현된 Chat 클래스를 사용하여 클라이언트에서 관찰자 참조를 만들고 구독한 후 메시지를 받도록 하는 것입니다.

코드는 다음과 유사합니다.

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

이제 서버의 조직에서 SendUpdateMessage 메서드를 호출할 때마다 구독된 모든 클라이언트가 메시지를 받습니다. 클라이언트 코드에서 변수 c의 Chat 인스턴스는 메시지를 수신하고 콘솔에 출력합니다.

중요

CreateObjectReference에 전달된 개체는 WeakReference<T>를 통해 유지되므로 다른 참조가 없는 경우 가비지 수집됩니다.

사용자는 수집하지 않으려는 각 관찰자에 대한 참조를 유지해야 합니다.

참고 항목

관찰자를 호스트하는 클라이언트가 실패할 수 있고 복구 후에 생성된 관찰자의 ID가 다르기(임의) 때문에 관찰자는 본질적으로 신뢰할 수 없습니다. ObserverManager<TObserver>는 위에서 설명한 대로 관찰자의 정기적인 재구독에 의존하여 비활성 관찰자를 제거할 수 있도록 합니다.

실행 모델

IGrainObserver의 구현은 IGrainFactory.CreateObjectReference 호출을 통해 등록되고 해당 메서드에 대한 각 호출은 해당 구현을 가리키는 새 참조를 만듭니다. Orleans는 이러한 각 참조에 하나씩 전송된 요청을 실행하여 완료합니다. 관찰자는 재진입되지 않으므로 관찰자에 대한 동시 요청은 Orleans에 의해 인터리브되지 않습니다. 동시에 요청을 수신하는 관찰자가 여러 개 있는 경우 해당 요청을 병렬로 실행할 수 있습니다. 관찰자 메서드의 실행은 AlwaysInterleaveAttribute 또는 ReentrantAttribute와 같은 특성에 영향을 받지 않습니다. 이러한 실행 모델은 개발자가 사용자 지정할 수 없습니다.