La mia origine dati invia informazioni in 100 lotti di 100 record con un ritardo compreso tra 1 e 3 secondi tra i batch.
Vorrei iniziare a elaborare i dati non appena viene ricevuto, ma non sono sicuro di come affrontarli al meglio.
Alcune idee con cui ho giocato includono:
yield Come puoi vedere, sono al settimo posto e apprezzerei alcune indicazioni testate su come avvicinarti a questo
Vorrei esaminare le API TPL Dataflow per questo tipo di lavoro. Al minimo si farebbe qualcosa di semplice come:
ActionBlock<MyData> myActionBlock = new ActionBlock<MyData>(data =>
{
// your processing logic here
});
E poi quando i tuoi messaggi arrivano da qualunque evento esterno, li hai appena postati in quel blocco:
myActionBlock.Post(data);
Questo è tutto. TPL Dataflow implementa quindi tutta la scala produttore / consumatore per voi coltivando il lavoro che avete fornito nella vostra definizione di azione sul numero di thread di lavoro possibili che la macchina può gestire (che si basa su tutte le solite euristiche della TPL).
Da lì puoi iniziare a diventare un po 'più avanzato semplicemente prendendo il controllo di governare esattamente quanta capacità / scala desideri con ExecutionDataflowBlockOptions . Puoi anche inserire una CancelationToken tramite EDBO.
Quindi puoi iniziare ad essere veramente avanzato usando le funzionalità di composizione dei diversi tipi di "blocchi" di TPL Dataflow e fare cose come introdurre un BufferBlock che, nel tuo caso, ti consentirebbe di ricevere quell'esplosione di 100 voci e non bloccare il produttore che sta effettuando la chiamata di .Post . Solo se raggiungi 100, il produttore non può più aggiungere. C'è anche BatchBlock che, a seconda del tuo carico di lavoro esatto, potrebbe avere più senso usare in modo che tu possa inviare 20 elementi a un thread di lavoro per essere elaborati anziché 1 alla volta.
Puoi collegare tra loro qualsiasi miscela di blocchi come puoi immaginare. C'è il filtraggio, la trasformazione ... lo dai il nome. Per coronare il tutto è anche possibile scrivere i propri blocchi personalizzati. Ad esempio, se si desidera introdurre l'accoppiamento temporale nel mix, è possibile, ad esempio, eseguire il batch fino a 20 item o se non si ricevono 20 per 10 che si inviano attraverso comunque molti già ricevuti . Le possibilità sono praticamente illimitate.
tornare alle basi con un buffer produttore-consumatore (guarda le code di sicurezza dei thread)
il tuo codice ricevente invia ogni record a una coda
quindi diversi thread di elaborazione estraggono un record e lo elaborano in un ciclo
Leggi altre domande sui tag c# abstraction data-structures concurrency multithreading