TPL 데이터 흐름 파이프 지속적으로 서비스를 예를 들어

Question

내에서 실행, 나는 3 개 블록을 가지고있다. 액션 블록에서 Completion()을 호출하지 않고 영원히 계속 실행되는 파이프를 생성하고 전체 파이프를 중지하는 방법.

Answer 1

응용 프로그램의 수명 동안 파이프 라인을 유지해야하는 경우 요청을 유지하는 데 정적 클래스를 사용할 수 있습니다. 액션 블록에서 complete를 호출 할 필요는 없습니다. 필요에 따라 다른 옵션은 응용 프로그램과 처리 파이프 라인을 분리하는 것입니다. 이들은 데이터베이스 메시지 대기열 또는 별도의 서버 측 응용 프로그램으로 구분할 수 있습니다.

@svick은 TaskCompletionSource를 사용하여 파이프 라인이 특정 항목으로 끝났을 때를 확인하는 것이 좋습니다. 여기 모두 함께 두는 것은 도움이 될 수있는 빠른 샘플의 : 파이프 라인 내에서 특정 작업의 완료를 조정하는

public class Controller { 

    public async Task<int> PostToPipeline(int inputValue) { 
     var message = new MessageIn(inputValue); 
     MyPipeline.InputBuffer.Post(message); 
     return await message.Completion.Task; 
    } 
} 

public class MessageIn { 
    public MessageIn(int value) { 
     InputValue = value; 
     Completion = new TaskCompletionSource<int>(); 
    } 

    public int InputValue { get; set; } 
    public TaskCompletionSource<int> Completion { get; set; } 
} 

public class MessageProcessed { 
    public int ProcessedValue { get; set; } 
    public TaskCompletionSource<int> Completion { get; set; } 
} 

public static class MyPipeline { 

    public static BufferBlock<MessageIn> InputBuffer { get; private set; } 
    private static TransformBlock<MessageIn, MessageProcessed> transform; 
    private static ActionBlock<MessageProcessed> action; 

    static MyPipeline() { 
     BuildPipeline(); 
     LinkPipeline(); 

    } 

    static void BuildPipeline() { 
     InputBuffer = new BufferBlock<MessageIn>(); 

     transform = new TransformBlock<MessageIn, MessageProcessed>((Func<MessageIn, MessageProcessed>)TransformMessage, new ExecutionDataflowBlockOptions() { 
      MaxDegreeOfParallelism = Environment.ProcessorCount, 
      BoundedCapacity = 10 
     }); 

     action = new ActionBlock<MessageProcessed>((Action<MessageProcessed>)CompletedProcessing, new ExecutionDataflowBlockOptions() { 
      MaxDegreeOfParallelism = Environment.ProcessorCount, 
      BoundedCapacity = 10 
     }); 
    } 

    static void LinkPipeline() { 
     InputBuffer.LinkTo(transform, new DataflowLinkOptions() { PropagateCompletion = true }); 
     transform.LinkTo(action, new DataflowLinkOptions() { PropagateCompletion = true }); 
    } 

    static MessageProcessed TransformMessage(MessageIn message) { 
     return new MessageProcessed() { 
      ProcessedValue = message.InputValue++, 
      Completion = message.Completion 
     }; 
    } 

    static void CompletedProcessing(MessageProcessed message) { 
     message.Completion.SetResult(message.ProcessedValue); 
    } 
} 

이 몇 가지 방법; 완료 소스를 기다리는 것이 귀하의 요구에 가장 적합한 방법 일 수 있습니다.

Answer 2

Dataflow는 단순한 파이프 라인 이상의 기능을 지원하기 때문에 특정 입력에 대한 파이프 라인 출력을 얻는 훌륭한 솔루션이 없습니다.

해결 방법은 TaskCompletionSource<T>을 만들고 파이프 라인에 입력과 함께 보내면 해결할 수 있습니다. 파이프 라인의 각 블록은이를 다음 블록으로 보내고 마지막 블록은 SetResult()을 호출합니다.

입력을 파이프 라인에 보내는 코드는 awaitTaskCompletionSource의 Task 파이프 라인의 출력을 기다릴 수 있습니다.