A Reliable Services kommunikációs API-k használata

Cikk
06/01/2023

Az Azure Service Fabric mint platform teljesen agnosztikus a szolgáltatások közötti kommunikációval kapcsolatban. Minden protokoll és verem elfogadható az UDP-től a HTTP-ig. A szolgáltatásfejlesztőn múlik, hogy a szolgáltatások hogyan kommunikáljanak. A Reliable Services alkalmazás-keretrendszer beépített kommunikációs vermeket és API-kat biztosít, amelyekkel egyéni kommunikációs összetevőket hozhat létre.

Szolgáltatáskommunikáció beállítása

A Reliable Services API egy egyszerű felületet használ a szolgáltatáskommunikációhoz. A szolgáltatás végpontjának megnyitásához egyszerűen implementálja ezt a felületet:


public interface ICommunicationListener
{
    Task<string> OpenAsync(CancellationToken cancellationToken);

    Task CloseAsync(CancellationToken cancellationToken);

    void Abort();
}

public interface CommunicationListener {
    CompletableFuture<String> openAsync(CancellationToken cancellationToken);

    CompletableFuture<?> closeAsync(CancellationToken cancellationToken);

    void abort();
}

Ezután hozzáadhatja a kommunikációs figyelő implementációját egy szolgáltatásalapú osztálymetódus-felülbírálással.

Állapot nélküli szolgáltatások esetén:

public class MyStatelessService : StatelessService
{
    protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
    {
        ...
    }
    ...
}

public class MyStatelessService extends StatelessService {

    @Override
    protected List<ServiceInstanceListener> createServiceInstanceListeners() {
        ...
    }
    ...
}

Állapotalapú szolgáltatások esetén:

    @Override
    protected List<ServiceReplicaListener> createServiceReplicaListeners() {
        ...
    }
    ...

public class MyStatefulService : StatefulService
{
    protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
    {
        ...
    }
    ...
}

Mindkét esetben figyelőgyűjteményt ad vissza. Ha több figyelőt használ, a szolgáltatás több végponton is figyelhet, akár különböző protokollokat is használva. Lehet például, hogy rendelkezik EGY HTTP-figyelővel és egy külön WebSocket-figyelővel. Ha nem biztonságos figyelővel és biztonságos figyelővel is rendelkezik, mindkét forgatókönyvet először engedélyezheti a nem biztonságos figyelőről a biztonságos újraegyenlítésre. Minden figyelő kap egy nevet, és az eredményül kapott névgyűjtemény: a címpárok JSON-objektumként jelennek meg, amikor egy ügyfél egy szolgáltatáspéldány vagy egy partíció figyelési címeit kéri le.

Állapot nélküli szolgáltatásban a felülbírálás a ServiceInstanceListeners gyűjteményét adja vissza. Az A ServiceInstanceListener egy függvényt tartalmaz, amely létrehoz egy függvényt ICommunicationListener(C#) / CommunicationListener(Java) , és nevet ad neki. Állapotalapú szolgáltatások esetén a felülbírálás a ServiceReplicaListeners gyűjteményét adja vissza. Ez némileg eltér az állapot nélküli megfelelőjétől, mivel a ServiceReplicaListener másodlagos replikákon lehetősége van megnyitni egy ICommunicationListener példányt. Nem csak több kommunikációs figyelőt használhat egy szolgáltatásban, hanem azt is megadhatja, hogy mely figyelők fogadják el a kéréseket a másodlagos replikákon, és melyek csak az elsődleges replikákon figyelnek.

Létrehozhat például egy ServiceRemotingListenert, amely csak az elsődleges replikákon fogad RPC-hívásokat, valamint egy második, egyéni figyelőt, amely HTTP-n keresztül fogad olvasási kéréseket a másodlagos replikákon:

protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
{
    return new[]
    {
        new ServiceReplicaListener(context =>
            new MyCustomHttpListener(context),
            "HTTPReadonlyEndpoint",
            true),

        new ServiceReplicaListener(context =>
            this.CreateServiceRemotingListener(context),
            "rpcPrimaryEndpoint",
            false)
    };
}

Megjegyzés

Ha több figyelőt hoz létre egy szolgáltatáshoz, minden figyelőnek egyedi nevet kell adni.

Végül ismertesse a szolgáltatáshoz szükséges végpontokat a szolgáltatásjegyzékben a végpontok szakaszában.

<Resources>
    <Endpoints>
      <Endpoint Name="WebServiceEndpoint" Protocol="http" Port="80" />
      <Endpoint Name="OtherServiceEndpoint" Protocol="tcp" Port="8505" />
    <Endpoints>
</Resources>

A kommunikációs figyelő hozzáférhet a számára lefoglalt végpont-erőforrásokhoz a CodePackageActivationContextServiceContext(z) fájlban. A figyelő ezután megkezdheti a kérelmek figyelését a megnyitáskor.

var codePackageActivationContext = serviceContext.CodePackageActivationContext;
var port = codePackageActivationContext.GetEndpoint("ServiceEndpoint").Port;

CodePackageActivationContext codePackageActivationContext = serviceContext.getCodePackageActivationContext();
int port = codePackageActivationContext.getEndpoint("ServiceEndpoint").getPort();

Megjegyzés

A végponterőforrások a teljes szolgáltatáscsomagra jellemzőek, és a Service Fabric lefoglalja őket a szolgáltatáscsomag aktiválásakor. Az ugyanabban a ServiceHostban üzemeltetett több szolgáltatásreplika is osztozhat ugyanazon a porton. Ez azt jelenti, hogy a kommunikációs figyelőnek támogatnia kell a portmegosztást. Ennek ajánlott módja, hogy a kommunikációs figyelő használja a partícióazonosítót és a replika/példány azonosítóját a figyelőcím létrehozásakor.

Szolgáltatáscím-regisztráció

Az elnevezési szolgáltatás nevű rendszerszolgáltatás Service Fabric-fürtökön fut. Az elnevezési szolgáltatás a szolgáltatások regisztrálója, és azok címei, amelyeket a szolgáltatás minden példánya vagy replikája figyel. Amikor a OpenAsync(C#) / openAsync(Java) metódus befejeződött ICommunicationListener(C#) / CommunicationListener(Java) , a visszaadott érték regisztrálva lesz az elnevezési szolgáltatásban. Ez az elnevezési szolgáltatásban közzétett visszaadott érték egy olyan sztring, amelynek értéke bármi lehet. Ez a sztringérték az, amit az ügyfelek látnak, amikor a szolgáltatás címét kérik az elnevezési szolgáltatástól.

public Task<string> OpenAsync(CancellationToken cancellationToken)
{
    EndpointResourceDescription serviceEndpoint = serviceContext.CodePackageActivationContext.GetEndpoint("ServiceEndpoint");
    int port = serviceEndpoint.Port;

    this.listeningAddress = string.Format(
                CultureInfo.InvariantCulture,
                "http://+:{0}/",
                port);

    this.publishAddress = this.listeningAddress.Replace("+", FabricRuntime.GetNodeContext().IPAddressOrFQDN);

    this.webApp = WebApp.Start(this.listeningAddress, appBuilder => this.startup.Invoke(appBuilder));

    // the string returned here will be published in the Naming Service.
    return Task.FromResult(this.publishAddress);
}

public CompletableFuture<String> openAsync(CancellationToken cancellationToken)
{
    EndpointResourceDescription serviceEndpoint = serviceContext.getCodePackageActivationContext.getEndpoint("ServiceEndpoint");
    int port = serviceEndpoint.getPort();

    this.publishAddress = String.format("http://%s:%d/", FabricRuntime.getNodeContext().getIpAddressOrFQDN(), port);

    this.webApp = new WebApp(port);
    this.webApp.start();

    /* the string returned here will be published in the Naming Service.
     */
    return CompletableFuture.completedFuture(this.publishAddress);
}

A Service Fabric olyan API-kat biztosít, amelyek lehetővé teszik az ügyfelek és más szolgáltatások számára, hogy ezt a címet szolgáltatásnév alapján kérzék. Ez azért fontos, mert a szolgáltatás címe nem statikus. A szolgáltatások erőforrás-kiegyensúlyozási és rendelkezésre állási célokból kerülnek át a fürtbe. Ez az a mechanizmus, amely lehetővé teszi az ügyfelek számára egy szolgáltatás figyelési címének feloldását.

Megjegyzés

A kommunikációs figyelők írásának teljes körű megismeréséhez tekintse meg a Service Fabric Web API-szolgáltatásokat az OWIN-alapú önkiszolgáló C#-szolgáltatással , míg a Java esetében saját HTTP-kiszolgáló implementációt írhat, lásd az EchoServer-alkalmazás példáját a címen https://github.com/Azure-Samples/service-fabric-java-getting-started.

Kommunikáció egy szolgáltatással

A Reliable Services API a következő kódtárakat biztosítja a szolgáltatásokkal kommunikáló ügyfelek írásához.

Szolgáltatásvégpont felbontása

A szolgáltatással való kommunikáció első lépése a beszélni kívánt szolgáltatás partíciójának vagy példányának végpontcímének feloldása. A ServicePartitionResolver(C#) / FabricServicePartitionResolver(Java) segédprogramosztály egy alapszintű primitív, amely segít az ügyfeleknek meghatározni egy szolgáltatás végpontját futásidőben. A Service Fabric terminológiájában a szolgáltatás végpontjának meghatározására szolgáló folyamatot szolgáltatásvégpont-feloldásnak nevezzük.

A fürtön belüli szolgáltatásokhoz való csatlakozáshoz a ServicePartitionResolver az alapértelmezett beállításokkal hozható létre. Ez az ajánlott használat a legtöbb esetben:

ServicePartitionResolver resolver = ServicePartitionResolver.GetDefault();

FabricServicePartitionResolver resolver = FabricServicePartitionResolver.getDefault();

Ha egy másik fürt szolgáltatásaihoz szeretne csatlakozni, létrehozhat egy ServicePartitionResolver-et fürtátjáró-végpontok készletével. Vegye figyelembe, hogy az átjáróvégpontok csak különböző végpontok ugyanahhoz a fürthöz való csatlakozáshoz. Például:

ServicePartitionResolver resolver = new  ServicePartitionResolver("mycluster.cloudapp.azure.com:19000", "mycluster.cloudapp.azure.com:19001");

FabricServicePartitionResolver resolver = new  FabricServicePartitionResolver("mycluster.cloudapp.azure.com:19000", "mycluster.cloudapp.azure.com:19001");

ServicePartitionResolver Másik lehetőségként a belső használatra szolgáló függvényt FabricClient is megadhat:

public delegate FabricClient CreateFabricClientDelegate();

public FabricServicePartitionResolver(CreateFabricClient createFabricClient) {
...
}

public interface CreateFabricClient {
    public FabricClient getFabricClient();
}

FabricClient az az objektum, amely a Service Fabric-fürttel való kommunikációra szolgál a fürt különböző felügyeleti műveleteihez. Ez akkor hasznos, ha nagyobb mértékben szeretné szabályozni, hogy egy szolgáltatáspartíció-feloldó hogyan kommunikáljon a fürttel. FabricClient belső gyorsítótárazást végez, és általában költséges létrehozni, ezért fontos, hogy a lehető legnagyobb mértékben újra felhasználja FabricClient a példányokat.

ServicePartitionResolver resolver = new  ServicePartitionResolver(() => CreateMyFabricClient());

FabricServicePartitionResolver resolver = new  FabricServicePartitionResolver(() -> new CreateFabricClientImpl());

Ezután egy feloldási módszer használatával lekérheti egy szolgáltatás vagy egy szolgáltatáspartíció címét a particionált szolgáltatásokhoz.

ServicePartitionResolver resolver = ServicePartitionResolver.GetDefault();

ResolvedServicePartition partition =
    await resolver.ResolveAsync(new Uri("fabric:/MyApp/MyService"), new ServicePartitionKey(), cancellationToken);

FabricServicePartitionResolver resolver = FabricServicePartitionResolver.getDefault();

CompletableFuture<ResolvedServicePartition> partition =
    resolver.resolveAsync(new URI("fabric:/MyApp/MyService"), new ServicePartitionKey());

A szolgáltatáscímek egyszerűen feloldhatók ServicePartitionResolver használatával, de több munkára van szükség a feloldott cím helyes használatának biztosításához. Az ügyfélnek észlelnie kell, hogy a csatlakozási kísérlet átmeneti hiba miatt meghiúsult-e, és újra lehet-e próbálkozni (például a szolgáltatás áthelyezhető vagy ideiglenesen nem érhető el), vagy állandó hibát (például a szolgáltatást törölték, vagy a kért erőforrás már nem létezik). A szolgáltatáspéldányok vagy replikák több okból is bármikor áttérhetnek csomópontról csomópontra. A ServicePartitionResolver által feloldott szolgáltatáscím elavult lehet, mire az ügyfélkód megpróbál csatlakozni. Ebben az esetben az ügyfélnek újra fel kell oldania a címet. Az előző ResolvedServicePartition megadása azt jelzi, hogy a feloldónak újra kell próbálkoznia a gyorsítótárazott cím lekérése helyett.

Az ügyfélkódnak általában nem kell közvetlenül a ServicePartitionResolverrel dolgoznia. A szolgáltatás a Reliable Services API-ban jön létre és továbbítja a kommunikációs ügyfél-előállítóknak. A gyárak a feloldót belsőleg használják egy ügyfélobjektum létrehozásához, amely a szolgáltatásokkal való kommunikációhoz használható.

Kommunikációs ügyfelek és gyárak

A Communication Factory kódtára egy tipikus hibakezelési újrapróbálkozási mintát valósít meg, amely megkönnyíti a kapcsolatok újrapróbálkozását a feloldott szolgáltatásvégpontok érdekében. A gyári kódtár biztosítja az újrapróbálkozás mechanizmusát, miközben megadja a hibakezelőket.

ICommunicationClientFactory(C#) / CommunicationClientFactory(Java) Meghatározza a kommunikációs ügyfél-előállító által implementált alapfelületet, amely olyan ügyfeleket állít elő, amelyek képesek kommunikálni a Service Fabric szolgáltatással. A CommunicationClientFactory megvalósítása a Service Fabric szolgáltatás által használt kommunikációs veremtől függ, ahol az ügyfél kommunikálni szeretne. A Reliable Services API egy CommunicationClientFactoryBase<TCommunicationClient>. Ez a CommunicationClientFactory felület alapszintű implementációját biztosítja, és az összes kommunikációs verem közös feladatait hajtja végre. (Ezek a feladatok közé tartozik egy ServicePartitionResolver használata a szolgáltatásvégpont meghatározásához). Az ügyfelek általában az absztrakt CommunicationClientFactoryBase osztályt implementálják a kommunikációs veremre jellemző logika kezelésére.

A kommunikációs ügyfél csak egy címet kap, és azt használja egy szolgáltatáshoz való csatlakozáshoz. Az ügyfél bármilyen protokollt használhat.

public class MyCommunicationClient : ICommunicationClient
{
    public ResolvedServiceEndpoint Endpoint { get; set; }

    public string ListenerName { get; set; }

    public ResolvedServicePartition ResolvedServicePartition { get; set; }
}

public class MyCommunicationClient implements CommunicationClient {

    private ResolvedServicePartition resolvedServicePartition;
    private String listenerName;
    private ResolvedServiceEndpoint endPoint;

    /*
     * Getters and Setters
     */
}

Az ügyfél-előállító elsősorban kommunikációs ügyfelek létrehozásáért felelős. Azon ügyfelek esetében, amelyek nem tartanak fenn állandó kapcsolatot, például HTTP-ügyfelet, a gyárnak csak az ügyfelet kell létrehoznia és visszaadnia. Az állandó kapcsolatot fenntartó egyéb protokollokat, például néhány bináris protokollt a gyárnak is ellenőriznie kell (ValidateClient(string endpoint, MyCommunicationClient client)) annak megállapításához, hogy újra létre kell-e hozni a kapcsolatot.

public class MyCommunicationClientFactory : CommunicationClientFactoryBase<MyCommunicationClient>
{
    protected override void AbortClient(MyCommunicationClient client)
    {
    }

    protected override Task<MyCommunicationClient> CreateClientAsync(string endpoint, CancellationToken cancellationToken)
    {
    }

    protected override bool ValidateClient(MyCommunicationClient clientChannel)
    {
    }

    protected override bool ValidateClient(string endpoint, MyCommunicationClient client)
    {
    }
}

public class MyCommunicationClientFactory extends CommunicationClientFactoryBase<MyCommunicationClient> {

    @Override
    protected boolean validateClient(MyCommunicationClient clientChannel) {
    }

    @Override
    protected boolean validateClient(String endpoint, MyCommunicationClient client) {
    }

    @Override
    protected CompletableFuture<MyCommunicationClient> createClientAsync(String endpoint) {
    }

    @Override
    protected void abortClient(MyCommunicationClient client) {
    }
}

Végül a kivételkezelő felelős annak meghatározásáért, hogy milyen műveletet kell elvégezni a kivétel bekövetkezésekor. A kivételek újrapróbálkozhatók és nem újrapróbálkozhatók.

A nem újrapróbálható kivételek egyszerűen visszajuttatják a hívót.
az újrapróbálható kivételek további kategorizálása átmeneti és nem átmeneti.
- Átmeneti kivételek azok, amelyek egyszerűen újrapróbálhatók a szolgáltatásvégpont címének újbóli feloldása nélkül. Ezek közé tartoznak az átmeneti hálózati problémák vagy a szolgáltatáshiba-válaszok, amelyek nem jelzik, hogy a szolgáltatásvégpont címe nem létezik.
- A nem átmeneti kivételek azok, amelyek a szolgáltatásvégpont címének újbóli feloldását igénylik. Ezek közé tartoznak azok a kivételek, amelyek azt jelzik, hogy a szolgáltatásvégpont nem érhető el, ami azt jelzi, hogy a szolgáltatás egy másik csomópontra lett áthelyezve.

A TryHandleException döntés egy adott kivételről. Ha nem tudja, milyen döntéseket kell hoznia egy kivételről, akkor hamis értéket kell visszaadnia. Ha tudja, hogy milyen döntést kell hoznia, ennek megfelelően kell beállítania az eredményt, és igaz értéket kell visszaadnia.

class MyExceptionHandler : IExceptionHandler
{
    public bool TryHandleException(ExceptionInformation exceptionInformation, OperationRetrySettings retrySettings, out ExceptionHandlingResult result)
    {
        // if exceptionInformation.Exception is known and is transient (can be retried without re-resolving)
        result = new ExceptionHandlingRetryResult(exceptionInformation.Exception, true, retrySettings, retrySettings.DefaultMaxRetryCount);
        return true;


        // if exceptionInformation.Exception is known and is not transient (indicates a new service endpoint address must be resolved)
        result = new ExceptionHandlingRetryResult(exceptionInformation.Exception, false, retrySettings, retrySettings.DefaultMaxRetryCount);
        return true;

        // if exceptionInformation.Exception is unknown (let the next IExceptionHandler attempt to handle it)
        result = null;
        return false;
    }
}

public class MyExceptionHandler implements ExceptionHandler {

    @Override
    public ExceptionHandlingResult handleException(ExceptionInformation exceptionInformation, OperationRetrySettings retrySettings) {

        /* if exceptionInformation.getException() is known and is transient (can be retried without re-resolving)
         */
        result = new ExceptionHandlingRetryResult(exceptionInformation.getException(), true, retrySettings, retrySettings.getDefaultMaxRetryCount());
        return true;


        /* if exceptionInformation.getException() is known and is not transient (indicates a new service endpoint address must be resolved)
         */
        result = new ExceptionHandlingRetryResult(exceptionInformation.getException(), false, retrySettings, retrySettings.getDefaultMaxRetryCount());
        return true;

        /* if exceptionInformation.getException() is unknown (let the next ExceptionHandler attempt to handle it)
         */
        result = null;
        return false;

    }
}

Végső összeállítás

A ICommunicationClient(C#) / CommunicationClient(Java), , és IExceptionHandler(C#) / ExceptionHandler(Java) egy kommunikációs protokoll köré épülve egy ServicePartitionClient(C#) / FabricServicePartitionClient(Java) sorba rendezi az egészet, és biztosítja a hibakezelési és szolgáltatáspartíciós címfeloldási hurkot ICommunicationClientFactory(C#) / CommunicationClientFactory(Java)ezen összetevők köré.

private MyCommunicationClientFactory myCommunicationClientFactory;
private Uri myServiceUri;

var myServicePartitionClient = new ServicePartitionClient<MyCommunicationClient>(
    this.myCommunicationClientFactory,
    this.myServiceUri,
    myPartitionKey);

var result = await myServicePartitionClient.InvokeWithRetryAsync(async (client) =>
   {
      // Communicate with the service using the client.
   },
   CancellationToken.None);

private MyCommunicationClientFactory myCommunicationClientFactory;
private URI myServiceUri;

FabricServicePartitionClient myServicePartitionClient = new FabricServicePartitionClient<MyCommunicationClient>(
    this.myCommunicationClientFactory,
    this.myServiceUri,
    myPartitionKey);

CompletableFuture<?> result = myServicePartitionClient.invokeWithRetryAsync(client -> {
      /* Communicate with the service using the client.
       */
   });

Megosztás a következőn keresztül:

A Reliable Services kommunikációs API-k használata

Szolgáltatáskommunikáció beállítása

Szolgáltatáscím-regisztráció

Kommunikáció egy szolgáltatással

Szolgáltatásvégpont felbontása

Kommunikációs ügyfelek és gyárak

Végső összeállítás

Következő lépések

Visszajelzés

Visszajelzés

További források