Så här använder du API:erna för Reliable Services-kommunikation
Azure Service Fabric som en plattform är helt oberoende av kommunikationen mellan tjänster. Alla protokoll och staplar är godtagbara, från UDP till HTTP. Det är upp till tjänstutvecklaren att välja hur tjänsterna ska kommunicera. Reliable Services-programramverket innehåller inbyggda kommunikationsstackar samt API:er som du kan använda för att skapa dina anpassade kommunikationskomponenter.
Konfigurera tjänstkommunikation
Reliable Services-API:et använder ett enkelt gränssnitt för tjänstkommunikation. Om du vill öppna en slutpunkt för din tjänst implementerar du bara det här gränssnittet:
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();
}
Du kan sedan lägga till implementeringen av kommunikationslyssnaren genom att returnera den i en tjänstbaserad åsidosättning av klassmetod.
För tillståndslösa tjänster:
public class MyStatelessService : StatelessService
{
protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
{
...
}
...
}
public class MyStatelessService extends StatelessService {
@Override
protected List<ServiceInstanceListener> createServiceInstanceListeners() {
...
}
...
}
För tillståndskänsliga tjänster:
@Override
protected List<ServiceReplicaListener> createServiceReplicaListeners() {
...
}
...
public class MyStatefulService : StatefulService
{
protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
{
...
}
...
}
I båda fallen returnerar du en samling lyssnare. Om du använder flera lyssnare kan tjänsten lyssna på flera slutpunkter, eventuellt med olika protokoll. Du kan till exempel ha en HTTP-lyssnare och en separat WebSocket-lyssnare. Du kan migrera från oskyddad till säker fjärrkommunikation genom att först aktivera båda scenarierna genom att ha både en icke-säker lyssnare och en säker lyssnare. Varje lyssnare får ett namn och den resulterande samlingen med namn : adresspar representeras som ett JSON-objekt när en klient begär lyssningsadresserna för en tjänstinstans eller en partition.
I en tillståndslös tjänst returnerar åsidosättningen en samling ServiceInstanceListeners. En ServiceInstanceListener innehåller en funktion för att skapa en ICommunicationListener(C#) / CommunicationListener(Java) och ger den ett namn. För tillståndskänsliga tjänster returnerar åsidosättningen en samling ServiceReplicaListeners. Detta skiljer sig något från dess tillståndslösa motsvarighet eftersom en ServiceReplicaListener har ett alternativ för att öppna en ICommunicationListener på sekundära repliker. Du kan inte bara använda flera kommunikationslyssnare i en tjänst, utan du kan också ange vilka lyssnare som accepterar begäranden på sekundära repliker och vilka som bara lyssnar på primära repliker.
Du kan till exempel ha en ServiceRemotingListener som endast tar RPC-anrop på primära repliker och en andra anpassad lyssnare som tar läsbegäranden på sekundära repliker via HTTP:
protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
{
return new[]
{
new ServiceReplicaListener(context =>
new MyCustomHttpListener(context),
"HTTPReadonlyEndpoint",
true),
new ServiceReplicaListener(context =>
this.CreateServiceRemotingListener(context),
"rpcPrimaryEndpoint",
false)
};
}
Anteckning
När du skapar flera lyssnare för en tjänst måste varje lyssnare få ett unikt namn.
Beskriv slutligen de slutpunkter som krävs för tjänsten i tjänstmanifestet under avsnittet på slutpunkter.
<Resources>
<Endpoints>
<Endpoint Name="WebServiceEndpoint" Protocol="http" Port="80" />
<Endpoint Name="OtherServiceEndpoint" Protocol="tcp" Port="8505" />
<Endpoints>
</Resources>
Kommunikationslyssnaren kan komma åt slutpunktsresurserna som allokerats till den från CodePackageActivationContext i .ServiceContext Lyssnaren kan sedan börja lyssna efter begäranden när den öppnas.
var codePackageActivationContext = serviceContext.CodePackageActivationContext;
var port = codePackageActivationContext.GetEndpoint("ServiceEndpoint").Port;
CodePackageActivationContext codePackageActivationContext = serviceContext.getCodePackageActivationContext();
int port = codePackageActivationContext.getEndpoint("ServiceEndpoint").getPort();
Anteckning
Slutpunktsresurser är gemensamma för hela tjänstpaketet och de allokeras av Service Fabric när tjänstpaketet aktiveras. Flera tjänstrepliker i samma ServiceHost kan dela samma port. Det innebär att kommunikationslyssnaren ska ha stöd för portdelning. Det rekommenderade sättet att göra detta är att kommunikationslyssnaren använder partitions-ID och replik-/instans-ID när lyssningsadressen genereras.
Registrering av tjänstadress
En systemtjänst som kallas namngivningstjänsten körs på Service Fabric-kluster. Namngivningstjänsten är en registrator för tjänster och deras adresser som varje instans eller replik av tjänsten lyssnar på. OpenAsync(C#) / openAsync(Java) När metoden för en ICommunicationListener(C#) / CommunicationListener(Java) slutförs registreras dess returvärde i namngivningstjänsten. Det här returvärdet som publiceras i namngivningstjänsten är en sträng vars värde kan vara vad som helst. Det här strängvärdet är vad klienter ser när de ber om en adress för tjänsten från namngivningstjänsten.
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);
}
Service Fabric tillhandahåller API:er som gör att klienter och andra tjänster sedan kan fråga efter den här adressen efter tjänstnamn. Detta är viktigt eftersom tjänstadressen inte är statisk. Tjänster flyttas runt i klustret för resursutjämning och tillgänglighet. Det här är den mekanism som gör det möjligt för klienter att matcha lyssningsadressen för en tjänst.
Anteckning
En fullständig genomgång av hur du skriver en kommunikationslyssnare finns i Service Fabric Web API-tjänster med OWIN-självvärd för C#, medan du kan skriva en egen HTTP-serverimplementering för Java i EchoServer-programexemplet på https://github.com/Azure-Samples/service-fabric-java-getting-started.
Kommunicera med en tjänst
Reliable Services-API:et tillhandahåller följande bibliotek för att skriva klienter som kommunicerar med tjänster.
Lösning för tjänstslutpunkt
Det första steget för kommunikation med en tjänst är att lösa en slutpunktsadress för partitionen eller instansen av tjänsten som du vill kommunicera med. Verktygsklassen ServicePartitionResolver(C#) / FabricServicePartitionResolver(Java) är en grundläggande primitiv som hjälper klienter att fastställa slutpunkten för en tjänst vid körning. I Service Fabric-terminologi kallas processen för att fastställa slutpunkten för en tjänst för lösning av tjänstslutpunkter.
Om du vill ansluta till tjänster i ett kluster kan ServicePartitionResolver skapas med standardinställningarna. Detta är den rekommenderade användningen för de flesta situationer:
ServicePartitionResolver resolver = ServicePartitionResolver.GetDefault();
FabricServicePartitionResolver resolver = FabricServicePartitionResolver.getDefault();
Om du vill ansluta till tjänster i ett annat kluster kan en ServicePartitionResolver skapas med en uppsättning klustergatewayslutpunkter. Observera att gatewayslutpunkter bara är olika slutpunkter för att ansluta till samma kluster. Exempel:
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 Du kan också få en funktion för att skapa en FabricClient som ska användas internt:
public delegate FabricClient CreateFabricClientDelegate();
public FabricServicePartitionResolver(CreateFabricClient createFabricClient) {
...
}
public interface CreateFabricClient {
public FabricClient getFabricClient();
}
FabricClient är det objekt som används för att kommunicera med Service Fabric-klustret för olika hanteringsåtgärder i klustret. Detta är användbart när du vill ha mer kontroll över hur en lösning för tjänstpartition interagerar med klustret. FabricClient utför cachelagring internt och är vanligtvis dyrt att skapa, så det är viktigt att återanvända FabricClient instanser så mycket som möjligt.
ServicePartitionResolver resolver = new ServicePartitionResolver(() => CreateMyFabricClient());
FabricServicePartitionResolver resolver = new FabricServicePartitionResolver(() -> new CreateFabricClientImpl());
En lösningsmetod används sedan för att hämta adressen till en tjänst eller en tjänstpartition för partitionerade tjänster.
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());
En tjänstadress kan enkelt lösas med hjälp av en ServicePartitionResolver, men mer arbete krävs för att säkerställa att den matchade adressen kan användas korrekt. Klienten måste identifiera om anslutningsförsöket misslyckades på grund av ett tillfälligt fel och kan försökas igen (t.ex. att tjänsten har flyttats eller är tillfälligt otillgänglig) eller ett permanent fel (t.ex. att tjänsten har tagits bort eller att den begärda resursen inte längre finns). Tjänstinstanser eller repliker kan när som helst flyttas från nod till nod av flera orsaker. Tjänstadressen som löses via ServicePartitionResolver kan vara inaktuell när klientkoden försöker ansluta. I så fall måste klienten matcha adressen igen. Om du anger föregående ResolvedServicePartition indikerar det att matcharen måste försöka igen i stället för att bara hämta en cachelagrad adress.
Vanligtvis behöver klientkoden inte fungera direkt med ServicePartitionResolver. Den skapas och skickas vidare till klientfabriker för kommunikation i Reliable Services-API:et. Fabrikerna använder matcharen internt för att generera ett klientobjekt som kan användas för att kommunicera med tjänster.
Kommunikationsklienter och fabriker
Kommunikationsfabriksbiblioteket implementerar ett typiskt mönster för felhantering av återförsök som gör det enklare att försöka ansluta igen till lösta tjänstslutpunkter. Fabriksbiblioteket tillhandahåller återförsöksmekanismen medan du anger felhanterare.
ICommunicationClientFactory(C#) / CommunicationClientFactory(Java) definierar basgränssnittet som implementeras av en kommunikationsklientfabrik som producerar klienter som kan kommunicera med en Service Fabric-tjänst. Implementeringen av CommunicationClientFactory beror på den kommunikationsstack som används av Service Fabric-tjänsten där klienten vill kommunicera. Reliable Services-API:et tillhandahåller en CommunicationClientFactoryBase<TCommunicationClient>. Detta ger en grundläggande implementering av CommunicationClientFactory-gränssnittet och utför uppgifter som är gemensamma för alla kommunikationsstackar. (Dessa uppgifter omfattar att använda en ServicePartitionResolver för att fastställa tjänstslutpunkten). Klienter implementerar vanligtvis den abstrakta Klassen CommunicationClientFactoryBase för att hantera logik som är specifik för kommunikationsstacken.
Kommunikationsklienten tar bara emot en adress och använder den för att ansluta till en tjänst. Klienten kan använda vilket protokoll den vill.
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
*/
}
Klientfabriken ansvarar främst för att skapa kommunikationsklienter. För klienter som inte upprätthåller en beständig anslutning, till exempel en HTTP-klient, behöver fabriken bara skapa och returnera klienten. Andra protokoll som upprätthåller en beständig anslutning, till exempel vissa binära protokoll, bör också verifieras (ValidateClient(string endpoint, MyCommunicationClient client)) av fabriken för att avgöra om anslutningen behöver återskapas.
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) {
}
}
Slutligen ansvarar en undantagshanterare för att avgöra vilken åtgärd som ska vidtas när ett undantag inträffar. Undantag kategoriseras i återförsöksbara och kan inte försökas igen.
- Icke-återförsöksbara undantag återförs helt enkelt till anroparen.
- återförsöksbara undantag kategoriseras ytterligare i tillfälliga och icke-tillfälliga.
- Tillfälliga undantag är de som helt enkelt kan göras om utan att matcha tjänstslutpunktsadressen på nytt. Dessa inkluderar tillfälliga nätverksproblem eller andra svar på tjänstfel än de som indikerar att tjänstslutpunktsadressen inte finns.
- Icke-tillfälliga undantag är de som kräver att tjänstslutpunktsadressen matchas på nytt. Dessa inkluderar undantag som anger att tjänstslutpunkten inte kunde nås, vilket indikerar att tjänsten har flyttats till en annan nod.
TryHandleException fattar ett beslut om ett visst undantag. Om den inte vet vilka beslut som ska fattas om ett undantag bör det returnera falskt. Om den vet vilket beslut som ska fattas bör den ange resultatet i enlighet med detta och returnera sant.
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;
}
}
Färdigställa allt
Med ett ICommunicationClient(C#) / CommunicationClient(Java), ICommunicationClientFactory(C#) / CommunicationClientFactory(Java)och IExceptionHandler(C#) / ExceptionHandler(Java) byggt runt ett kommunikationsprotokoll, omsluter en ServicePartitionClient(C#) / FabricServicePartitionClient(Java) allt och ger en loop för felhantering och adressmatchning av tjänstpartitioner runt dessa komponenter.
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.
*/
});