Jak používat rozhraní API pro komunikaci Reliable Services
Azure Service Fabric jako platforma je zcela nezávislá na komunikaci mezi službami. Všechny protokoly a zásobníky jsou přijatelné, od UDP po HTTP. Záleží na vývojáři služeb, aby zvolil způsob komunikace služeb. Aplikační architektura Reliable Services poskytuje integrované komunikační zásobníky a rozhraní API, která můžete použít k sestavení vlastních komunikačních komponent.
Nastavení komunikace služby
Rozhraní API Reliable Services používá pro komunikaci služby jednoduché rozhraní. Pokud chcete otevřít koncový bod pro vaši službu, jednoduše implementujte toto rozhraní:
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();
}
Pak můžete přidat implementaci naslouchacího procesu komunikace tak, že ji vrátíte v přepsání metody založené na službě.
Pro bezstavové služby:
public class MyStatelessService : StatelessService
{
protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
{
...
}
...
}
public class MyStatelessService extends StatelessService {
@Override
protected List<ServiceInstanceListener> createServiceInstanceListeners() {
...
}
...
}
Stavové služby:
@Override
protected List<ServiceReplicaListener> createServiceReplicaListeners() {
...
}
...
public class MyStatefulService : StatefulService
{
protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
{
...
}
...
}
V obou případech vrátíte kolekci naslouchacích procesů. Použití více naslouchacích procesů umožňuje službě naslouchat na více koncových bodech, potenciálně pomocí různých protokolů. Můžete mít například naslouchací proces HTTP a samostatný naslouchací proces WebSocket. Můžete migrovat z nezabezpečené na zabezpečenou vzdálenou komunikace tak, že nejprve povolíte oba scénáře tím, že budete mít nezabezpečený naslouchací proces i zabezpečený naslouchací proces. Každý naslouchací proces získá název a výsledná kolekce párů adres name : jsou reprezentovány jako objekt JSON, když klient požádá o naslouchací adresy pro instanci služby nebo oddíl.
V bezstavové službě vrátí přepsání kolekci ServiceInstanceListeners. Obsahuje ServiceInstanceListener funkci, která ICommunicationListener(C#) / CommunicationListener(Java) vytvoří a pojmenuje ji. U stavových služeb vrátí přepsání kolekci ServiceReplicaListeners. To se mírně liší od bezstavového protějšku ServiceReplicaListener , protože má možnost otevřít objekt ICommunicationListener na sekundárních replikách. Ve službě můžete nejen použít více komunikačních naslouchacích procesů, ale můžete také určit, které naslouchací procesy přijímají požadavky na sekundární repliky a které naslouchají jenom primárním replikám.
Můžete mít například ServiceRemotingListener, který přijímá volání RPC pouze na primárních replikách, a druhý vlastní naslouchací proces, který přijímá požadavky na čtení na sekundárních replikách přes PROTOKOL HTTP:
protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
{
return new[]
{
new ServiceReplicaListener(context =>
new MyCustomHttpListener(context),
"HTTPReadonlyEndpoint",
true),
new ServiceReplicaListener(context =>
this.CreateServiceRemotingListener(context),
"rpcPrimaryEndpoint",
false)
};
}
Poznámka
Při vytváření více naslouchacích procesů pro službu musí mít každý naslouchací proces jedinečný název.
Nakonec popište koncové body, které služba vyžaduje, v manifestu služby v části věnované koncovým bodům.
<Resources>
<Endpoints>
<Endpoint Name="WebServiceEndpoint" Protocol="http" Port="80" />
<Endpoint Name="OtherServiceEndpoint" Protocol="tcp" Port="8505" />
<Endpoints>
</Resources>
Komunikační naslouchací proces má přístup k prostředkům koncového bodu, které mu byly přiděleny, z v CodePackageActivationContextServiceContext. Naslouchací proces pak může při otevření začít naslouchat žádostem.
var codePackageActivationContext = serviceContext.CodePackageActivationContext;
var port = codePackageActivationContext.GetEndpoint("ServiceEndpoint").Port;
CodePackageActivationContext codePackageActivationContext = serviceContext.getCodePackageActivationContext();
int port = codePackageActivationContext.getEndpoint("ServiceEndpoint").getPort();
Poznámka
Prostředky koncových bodů jsou společné pro celý balíček služby a Service Fabric je přiděluje při aktivaci balíčku služby. Několik replik služeb hostovaných ve stejném hostiteli ServiceHost může sdílet stejný port. To znamená, že komunikační naslouchací proces by měl podporovat sdílení portů. Doporučený způsob, jak to udělat, je, aby komunikační naslouchací proces při generování adresy naslouchání použil ID oddílu a ID repliky nebo instance.
Registrace adresy služby
Systémová služba s názvem Služba pojmenování běží na clusterech Service Fabric. Služba pojmenování je registrátor služeb a jejich adres, na nichž naslouchá každá instance nebo replika služby. Po dokončení OpenAsync(C#) / openAsync(Java) metody ICommunicationListener(C#) / CommunicationListener(Java) se její návratová hodnota zaregistruje ve službě Pojmenování. Tato návratová hodnota, která se publikuje ve službě Pojmenování, je řetězec, jehož hodnotou může být vůbec cokoli. Tato řetězcová hodnota je to, co klienti uvidí, když od služby Pojmenování požádají o adresu služby.
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 poskytuje rozhraní API, která klientům a dalším službám umožňují žádat o tuto adresu podle názvu služby. To je důležité, protože adresa služby není statická. Služby se v clusteru přesouvají za účelem vyrovnávání prostředků a zajištění dostupnosti. Jedná se o mechanismus, který umožňuje klientům přeložit adresu naslouchající služby.
Poznámka
Kompletní návod, jak napsat naslouchací proces komunikace, najdete v tématu Služby webového rozhraní API Service Fabric se samoobslužným hostováním OWIN pro C#, zatímco pro Javu můžete napsat vlastní implementaci serveru HTTP, viz Příklad aplikace EchoServer na adrese https://github.com/Azure-Samples/service-fabric-java-getting-started.
Komunikace se službou
Rozhraní API Reliable Services poskytuje následující knihovny pro zápis klientů, kteří komunikují se službami.
Překlad koncového bodu služby
Prvním krokem při komunikaci se službou je překlad adresy koncového bodu oddílu nebo instance služby, se kterou chcete komunikovat. Třída ServicePartitionResolver(C#) / FabricServicePartitionResolver(Java) nástroje je základní primitivo, které pomáhá klientům určit koncový bod služby za běhu. V terminologii Service Fabric se proces určení koncového bodu služby označuje jako překlad koncového bodu služby.
Pro připojení ke službám v rámci clusteru je možné vytvořit ServicePartitionResolver pomocí výchozího nastavení. Toto je doporučené použití ve většině situací:
ServicePartitionResolver resolver = ServicePartitionResolver.GetDefault();
FabricServicePartitionResolver resolver = FabricServicePartitionResolver.getDefault();
Pokud se chcete připojit ke službám v jiném clusteru, můžete vytvořit ServicePartitionResolver se sadou koncových bodů brány clusteru. Všimněte si, že koncové body brány jsou jenom různé koncové body pro připojení ke stejnému clusteru. Příklad:
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");
Alternativně ServicePartitionResolver můžete získat funkci pro vytvoření , která FabricClient se použije interně:
public delegate FabricClient CreateFabricClientDelegate();
public FabricServicePartitionResolver(CreateFabricClient createFabricClient) {
...
}
public interface CreateFabricClient {
public FabricClient getFabricClient();
}
FabricClient je objekt, který se používá ke komunikaci s clusterem Service Fabric pro různé operace správy v clusteru. To je užitečné, když chcete mít větší kontrolu nad tím, jak překladač oddílů služby komunikuje s vaším clusterem. FabricClient interně provádí ukládání do mezipaměti a jeho vytváření je obecně nákladné, proto je důležité instance co nejvíce používat FabricClient .
ServicePartitionResolver resolver = new ServicePartitionResolver(() => CreateMyFabricClient());
FabricServicePartitionResolver resolver = new FabricServicePartitionResolver(() -> new CreateFabricClientImpl());
Metoda resolve se pak použije k načtení adresy služby nebo oddílu služby pro dělené služby.
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());
Adresu služby lze snadno přeložit pomocí ServicePartitionResolver, ale je potřeba více práce, aby bylo možné přeloženou adresu použít správně. Váš klient musí zjistit, jestli pokus o připojení selhal kvůli přechodné chybě a je možné ho opakovat (např. služba se přesunula nebo je dočasně nedostupná) nebo jestli nedošlo k trvalé chybě (např. došlo k odstranění služby nebo požadovaný prostředek už neexistuje). Instance nebo repliky služeb se můžou kdykoli přesouvat z uzlu na uzel z několika důvodů. Adresa služby vyřešená prostřednictvím ServicePartitionResolver může být v době, kdy se váš klientský kód pokusí připojit, zastaralá. V takovém případě klient znovu musí adresu přeložit. Pokud zadáte předchozí ResolvedServicePartition hodnotu, znamená to, že překladač musí zkusit znovu místo pouhého načtení adresy uložené v mezipaměti.
Kód klienta obvykle nemusí pracovat přímo s ServicePartitionResolver. Vytvoří se a předá se komunikačním klientským továrnám v rozhraní API Reliable Services. Továrny používají překladač interně k vygenerování objektu klienta, který lze použít ke komunikaci se službami.
Komunikační klienti a továrny
Knihovna komunikační továrny implementuje typický model opakování zpracování chyb, který usnadňuje opakování připojení k vyřešeným koncovým bodům služby. Knihovna továrny poskytuje mechanismus opakování při poskytování obslužných rutin chyb.
ICommunicationClientFactory(C#) / CommunicationClientFactory(Java) definuje základní rozhraní implementované objektem pro vytváření komunikačních klientů, které vytváří klienty, kteří můžou komunikovat se službou Service Fabric. Implementace CommunicationClientFactory závisí na komunikačním zásobníku používaném službou Service Fabric, kde chce klient komunikovat. Rozhraní API Reliable Services poskytuje CommunicationClientFactoryBase<TCommunicationClient>. To poskytuje základní implementaci rozhraní CommunicationClientFactory a provádí úlohy, které jsou společné pro všechny komunikační zásobníky. (Mezi tyto úlohy patří použití ServicePartitionResolver k určení koncového bodu služby.) Klienti obvykle implementují abstraktní Třídu CommunicationClientFactoryBase pro zpracování logiky, která je specifická pro komunikační zásobník.
Komunikační klient pouze obdrží adresu a použije ji pro připojení ke službě. Klient může použít jakýkoli protokol, který chce.
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
*/
}
Objekt pro vytváření klientů je primárně zodpovědný za vytváření komunikačních klientů. Pro klienty, kteří neudržuje trvalé připojení, jako je klient HTTP, stačí, když továrna vytvoří a vrátí klienta. Další protokoly, které udržují trvalé připojení, jako jsou některé binární protokoly, by také měly být ověřeny (ValidateClient(string endpoint, MyCommunicationClient client)) objektem pro vytváření, aby bylo možné určit, jestli je potřeba znovu vytvořit připojení.
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) {
}
}
Nakonec obslužná rutina výjimky je zodpovědná za určení akce, která se má provést, když dojde k výjimce. Výjimky jsou kategorizovány na opakovatelné a neopakovatelné.
- Výjimky, které nelze opakovat , se jednoduše znovu zobrazí zpět volajícímu.
- opakovaně opakovatelné výjimky jsou dále rozdělené na přechodné a nepřechodné.
- Přechodné výjimky jsou ty, které je možné jednoduše opakovat bez opětovného překladu adresy koncového bodu služby. Ty budou zahrnovat přechodné problémy se sítí nebo jiné odpovědi na chyby služby než ty, které značí, že adresa koncového bodu služby neexistuje.
- Výjimky, které nejsou přechodné , jsou ty, které vyžadují, aby se adresa koncového bodu služby přeložila znovu. Patří mezi ně výjimky, které značí, že koncový bod služby není dostupný, což znamená, že se služba přesunula do jiného uzlu.
Rozhodne TryHandleException o dané výjimce. Pokud neví, jaká rozhodnutí o výjimce provést, měla by vrátit hodnotu false. Pokud ví , jaké rozhodnutí má udělat, měl by odpovídajícím způsobem nastavit výsledek a vrátit hodnotu true.
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;
}
}
Spojení všech součástí dohromady
S objekty ICommunicationClient(C#) / CommunicationClient(Java), ICommunicationClientFactory(C#) / CommunicationClientFactory(Java)a IExceptionHandler(C#) / ExceptionHandler(Java) sestavenými kolem komunikačního protokolu ServicePartitionClient(C#) / FabricServicePartitionClient(Java) to všechno zabalí do sebe a poskytuje smyčku zpracování chyb a řešení adres oddílů služby kolem těchto komponent.
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.
*/
});