비동기식 대기열 관리자

Question

C#에서 비동기 다중 서버 네트워크 응용 프로그램을 작성하는 동안 문제가 발생했습니다. 스레드 풀에 의해 처리되는 작업이 많으며 여기에는 네트워크 소켓에 쓰기 작업이 포함됩니다. 이것은 하나 이상의 스레드가 동시에 소켓에 ​​쓸 수 있고 보내는 메시지를 분리 할 수있는 경우를 허용합니다. 이 문제를 해결하기위한 나의 아이디어는 큐에 데이터가 추가 될 때마다 소켓이 소켓 시스템에 메시지를 쓸 수있는 큐 시스템을 구현하는 것이 었습니다.

내 문제는이 성격의 아키텍처를 둘러싼 내 머리를 감쌀 수 없다는 것입니다. 데이터가 큐에 추가 될 때마다 이벤트를 발생시키는 큐 객체가 있다고 가정합니다. 그런 다음 이벤트는 대기열에 보관중인 데이터를 씁니다.하지만 두 개의 스레드가 동시에 대기열에 추가되면 대기열이 스레드 안전성을 갖더라도 이벤트가 계속해서 발생하고 나는 똑같은 문제를 겪을 것이다. 그렇다면 어쩌면 다른 이벤트가 진행 중일 때 이벤트를 중단 할 수도 있지만, 일단 뮤텍스 또는 일부 이벤트에서 스레드를 차단하지 않고 첫 번째 이벤트가 끝나면 어떻게 계속 진행할 수 있습니까? "block nothing"아키텍처를 엄격하게 유지하려고 시도하지 않는다면 그렇게 어렵지 않을 것입니다.하지만이 특정 응용 프로그램에서는 스레드 풀 스레드가 자신의 일을 계속하도록 허용해야합니다.

아이디어가 있으십니까?

Answer 1

Porges와 비슷한 답변이지만 구현이 조금 다릅니다.

우선, 보통 전송할 바이트를 대기열에 넣지 않지만 전송 스레드에서 객체를 대상으로 분류하지만 맛이 좋을 것 같습니다. 그러나 더 큰 차이점은 ConcurrentQueues (BlockingCollection 외에도)의 사용에 있습니다. 그래서 나는

 BlockingCollection<Packet> sendQueue = new BlockingCollection<Packet>(new ConcurrentQueue<Packet>()); 
     while (true) 
     { 
      var packet = sendQueue.Take(); //this blocks if there are no items in the queue. 
      SendPacket(packet); //Send your packet here. 
     } 

유사한 코드로 끝날 것 키 - 걸릴 멀리 여기에이 코드를 루프 하나 개의 스레드를 가지고, 모든 다른 스레드는 스레드 안전한 방법으로 큐에 추가 할 수 있습니다 (BlockingCollection과 ConcurrentQueue 모두 스레드로부터 안전함)

내가 비슷한 질문에 대답 한 Processing a queue of items asynchronously in C#을보십시오.

Answer 2

C#은 잘 모르겠지만 이벤트가 발생하면 소켓 관리자가 대기열에서 당겨서 한 번에 하나씩 쓰게됩니다. 이미 트리거가 실행 중이면 트리거가 아무 것도하지 않고 큐에 아무 것도 없으면 중지됩니다.

두 번째 이벤트가 아무 작업도 수행하지 않기 때문에 두 스레드가 동시에 큐에 쓰는 문제를 해결합니다.

Answer 3

모든 작업자 스레드가 결과를 쓰는 스레드 안전 큐를 가질 수 있습니다. 그런 다음 큐를 폴링하고 기다리는 결과를 전송하는 다른 스레드가 있습니다.

Answer 4

당신은 하나의 쓰레드를 동 기적으로 소켓에 쓰고 그 쓰레드가 처리 할 큐에 쓰는 것이 필요합니다.

당신은 열심히 일을 할 수있는 차단 모음 (BlockingCollection<T>)를 사용할 수 있습니다 :

는

// somewhere there is a queue: 

BlockingCollection<byte[]> queue = new BlockingCollection<byte[]>(); 

// in socket-writing thread, read from the queue and send the messages: 

foreach (byte[] message in queue.GetConsumingEnumerable()) 
{ 
    // just an example... obviously you'd need error handling and stuff here 
    socket.Send(message); 
} 

// in the other threads, just enqueue messages to be sent: 

queue.Add(someMessage); 

BlockingCollection 모든 동기화를 처리합니다. 최대 대기열 길이 및 기타 재미있는 것을 적용 할 수도 있습니다.