.NET Framework
TPL Dataflow
Sök…
Anmärkningar
Bibliotek som används i exempel
System.Threading.Tasks.Dataflow
System.Threading.Tasks
System.Net.Http
System.Net
Skillnaden mellan Post och SendAsync
För att lägga till objekt i ett block kan du antingen använda Post
eller SendAsync
.
Post
kommer att försöka lägga till objektet synkront och returnera en bool
säger om det lyckades eller inte. Det kanske inte lyckas när till exempel ett block har nått sin BoundedCapcity
och inte har mer utrymme för nya objekt ännu. SendAsync
å andra sidan kommer att returnera en oavslutad Task<bool>
som du kan await
. Denna uppgift kommer att slutföras i framtiden med ett true
resultat när blocket rensade sin interna kön och kan acceptera fler objekt eller med ett false
resultat om det minskar permanent (t.ex. till följd av avbokning).
Skickar till en ActionBlock och väntar på att den är klar
// Create a block with an asynchronous action
var block = new ActionBlock<string>(async hostName =>
{
IPAddress[] ipAddresses = await Dns.GetHostAddressesAsync(hostName);
Console.WriteLine(ipAddresses[0]);
});
block.Post("google.com"); // Post items to the block's InputQueue for processing
block.Post("reddit.com");
block.Post("stackoverflow.com");
block.Complete(); // Tell the block to complete and stop accepting new items
await block.Completion; // Asynchronously wait until all items completed processingu
Länka block för att skapa en pipeline
var httpClient = new HttpClient();
// Create a block the accepts a uri and returns its contents as a string
var downloaderBlock = new TransformBlock<string, string>(
async uri => await httpClient.GetStringAsync(uri));
// Create a block that accepts the content and prints it to the console
var printerBlock = new ActionBlock<string>(
contents => Console.WriteLine(contents));
// Make the downloaderBlock complete the printerBlock when its completed.
var dataflowLinkOptions = new DataflowLinkOptions {PropagateCompletion = true};
// Link the block to create a pipeline
downloaderBlock.LinkTo(printerBlock, dataflowLinkOptions);
// Post urls to the first block which will pass their contents to the second one.
downloaderBlock.Post("http://youtube.com");
downloaderBlock.Post("http://github.com");
downloaderBlock.Post("http://twitter.com");
downloaderBlock.Complete(); // Completion will propagate to printerBlock
await printerBlock.Completion; // Only need to wait for the last block in the pipeline
Synkron tillverkare / konsument med BufferBlock
public class Producer
{
private static Random random = new Random((int)DateTime.UtcNow.Ticks);
//produce the value that will be posted to buffer block
public double Produce ( )
{
var value = random.NextDouble();
Console.WriteLine($"Producing value: {value}");
return value;
}
}
public class Consumer
{
//consume the value that will be received from buffer block
public void Consume (double value) => Console.WriteLine($"Consuming value: {value}");
}
class Program
{
private static BufferBlock<double> buffer = new BufferBlock<double>();
static void Main (string[] args)
{
//start a task that will every 1 second post a value from the producer to buffer block
var producerTask = Task.Run(async () =>
{
var producer = new Producer();
while(true)
{
buffer.Post(producer.Produce());
await Task.Delay(1000);
}
});
//start a task that will recieve values from bufferblock and consume it
var consumerTask = Task.Run(() =>
{
var consumer = new Consumer();
while(true)
{
consumer.Consume(buffer.Receive());
}
});
Task.WaitAll(new[] { producerTask, consumerTask });
}
}
Asynkron producentkonsument med en bunden buffertlock
var bufferBlock = new BufferBlock<int>(new DataflowBlockOptions
{
BoundedCapacity = 1000
});
var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token;
var producerTask = Task.Run(async () =>
{
var random = new Random();
while (!cancellationToken.IsCancellationRequested)
{
var value = random.Next();
await bufferBlock.SendAsync(value, cancellationToken);
}
});
var consumerTask = Task.Run(async () =>
{
while (await bufferBlock.OutputAvailableAsync())
{
var value = bufferBlock.Receive();
Console.WriteLine(value);
}
});
await Task.WhenAll(producerTask, consumerTask);
Modified text is an extract of the original Stack Overflow Documentation
Licensierat under CC BY-SA 3.0
Inte anslutet till Stack Overflow