Come fare Async nella funzione WebJob di Azure

Question

Ho un metodo asincrono che ottiene i dati API da un server. Quando eseguo questo codice sul mio computer locale, in un'app console, funziona ad alta velocità, passando attraverso alcune centinaia di chiamate http nella funzione asincrona al minuto. Quando metto lo stesso codice per essere attivato da un messaggio di coda WebJob di Azure, tuttavia, sembra funzionare in modo sincrono e i miei numeri scorrono - sono sicuro che mi manca qualcosa di semplice nel mio approccio - ogni assistenza è apprezzata.

(1) .. WebJob funzione che ascolta per un messaggio sulla coda e prende il via l'API ottenere il processo sul messaggio ricevuto:

public class Functions 
    { 
     // This function will get triggered/executed when a new message is written 
     // on an Azure Queue called queue. 

     public static async Task ProcessQueueMessage ([QueueTrigger("myqueue")] string message, TextWriter log) 
     { 
      var getAPIData = new GetData(); 
      getAPIData.DoIt(message).Wait(); 
      log.WriteLine("*** done: " + message); 
     } 
    } 

(2) la classe che le opere al di fuori azzurro in modalità asincrona a velocità ...

class GetData 
    { 
     // wrapper that is called by the message function trigger 
     public async Task DoIt(string MessageFile) 
     { 
      await CallAPI(MessageFile); 
     } 

     public async Task<string> CallAPI(string MessageFile) 
     { 
      /// create a list of sample APIs to call... 
      var apiCallList = new List<string>(); 
      apiCallList.Add("localhost/?q=1"); 
      apiCallList.Add("localhost/?q=2"); 
      apiCallList.Add("localhost/?q=3"); 
      apiCallList.Add("localhost/?q=4"); 
      apiCallList.Add("localhost/?q=5"); 

      // setup httpclient 
      HttpClient client = 
       new HttpClient() { MaxResponseContentBufferSize = 10000000 }; 
      var timeout = new TimeSpan(0, 5, 0); // 5 min timeout 
      client.Timeout = timeout; 

      // create a list of http api get Task... 
      IEnumerable<Task<string>> allResults = apiCallList.Select(str => ProcessURLPageAsync(str, client)); 
      // wait for them all to complete, then move on... 
      await Task.WhenAll(allResults); 

      return allResults.ToString(); 
     } 

     async Task<string> ProcessURLPageAsync(string APIAddressString, HttpClient client) 
     { 
      string page = ""; 
      HttpResponseMessage resX; 

      try 
      { 
       // set the address to call 
       Uri URL = new Uri(APIAddressString); 
       // execute the call 
       resX = await client.GetAsync(URL); 
       page = await resX.Content.ReadAsStringAsync(); 
       string rslt = page; 
       // do something with the api response data 
      } 
      catch (Exception ex) 
      { 
       // log error 
      } 
      return page; 
     } 

    } 

Answer 1

Innanzitutto perché la funzione attivata è async, è necessario utilizzare await anziché .Wait(). Wait bloccherà il thread corrente.

public static async Task ProcessQueueMessage([QueueTrigger("myqueue")] string message, TextWriter log) 
{ 
    var getAPIData = new GetData(); 
    await getAPIData.DoIt(message); 
    log.WriteLine("*** done: " + message); 
} 

In ogni caso sarete in grado di trovare informazioni utili dal documentation

esecuzione parallela

Se si dispone di più funzioni in ascolto su diverse code, l'SDK li chiamare parallelo quando i messaggi vengono ricevuti contemporaneamente.

Lo stesso è vero quando si ricevono più messaggi per una singola coda. Per impostazione predefinita, l'SDK ottiene un batch di 16 messaggi di coda alla volta ed esegue la funzione che li elabora in parallelo.The batch size is configurable. Quando il numero in elaborazione raggiunge la metà della dimensione del batch, l'SDK ottiene un altro batch e inizia a elaborare quei messaggi. Pertanto il numero massimo di messaggi simultanei elaborati per funzione è pari a una volta e mezza la dimensione del batch. Questo limite si applica separatamente a ciascuna funzione che ha un attributo QueueTrigger.

Ecco un codice di esempio per configurare la dimensione del lotto:

var config = new JobHostConfiguration(); 
config.Queues.BatchSize = 50; 
var host = new JobHost(config); 
host.RunAndBlock(); 

Tuttavia, non è sempre una buona opzione per avere troppi thread in esecuzione allo stesso tempo e potrebbe portare a cattive prestazioni.

Un'altra opzione è quella di scalare la vostra webjob:

Più istanze

se il vostro web app funziona su più istanze, un WebJob continuo viene eseguito su ogni macchina, e ogni macchina attenderà innesca e tenta di eseguire le funzioni. Il trigger della coda di WebJobs SDK impedisce automaticamente a una funzione di elaborare un messaggio di coda più volte; le funzioni non devono essere scritte per essere idempotenti. Tuttavia, se si desidera garantire che solo un'istanza di una funzione venga eseguita anche quando esistono più istanze dell'app Web host, è possibile utilizzare l'attributo Singleton.

Answer 2

Avere una lettura di questo Webjobs SDK documentation - il comportamento che si dovrebbe aspettare che il processo verrà eseguito ed elaborare un messaggio alla volta, ma sarà scalare se si creano più istanze (del vostro servizio app). Se avessi più code, si innescherebbero in parallelo.

Per migliorare le prestazioni, consultare la sezione delle impostazioni di configurazione nel collegamento che ho inviato, che si riferisce al numero di messaggi che possono essere attivati ​​in un batch.

Se si desidera elaborare più messaggi in parallelo e non si vuole fare affidamento sul ridimensionamento dell'istanza, è necessario utilizzare invece il threading (async non riguarda il parallelismo multi-thread, ma rende un uso più efficiente di il thread che stai utilizzando). Quindi la funzione di trigger della coda dovrebbe leggere il messaggio dalla coda, creare un thread e "attivare e disattivare" quel thread e quindi tornare dalla funzione trigger. Questo contrassegna il messaggio come elaborato e consente di elaborare il successivo messaggio in coda, anche se in teoria si sta ancora elaborando quello precedente. Nota che dovrai includere la tua logica per la gestione degli errori e assicurarti che i dati non vadano persi se il tuo thread genera un'eccezione o non può elaborare il messaggio (ad esempio mettilo su una coda velenosa).

L'altra opzione è di non utilizzare l'attributo [queuetrigger] e utilizzare direttamente le funzioni dell'API sdk delle code di archiviazione di Azure per connettersi ed elaborare i messaggi in base alle proprie esigenze.