블로킹 콜렉션의 기본 기본 동시 큐에 액세스하는 방법

Question

여러 개의 생성자와 단일 소비자가 있습니다. 그러나 대기열에 아직 소비되지 않은 항목이 있으면 생산자가 다시 대기열에 대기 시켜서는 안됩니다. 차단 콜은이 방법을 포함 않으며이 방법 같은 trypeek()의 ​​모든 종류를 제공 않습니다이없는 그러나

if (!myBlockingColl.Contains(item)) 
    myBlockingColl.Add(item) 

을 (독특한 없음은 기본 동시 큐를 사용하여 수집을 차단와 중복). 기본 동시 대기열에 액세스 할 수 있으므로 어떻게 할 수 있습니까?

if (!myBlockingColl.myConcurQ.trypeek(item) 
    myBlockingColl.Add(item) 

테일 스핀에서. 도와주세요. 덕분에

Answer 1

자물쇠로 작업을 구현하여 항목을 읽고 손상시키지 않도록 항목을 작성하여 원자화하는 것이 좋습니다. 예를 들어, 임의의 IEnumerable의 경우 :

object bcLocker = new object(); 

// ... 

lock (bcLocker) 
{ 
    bool foundTheItem = false; 
    foreach (someClass nextItem in myBlockingColl) 
    { 
     if (nextItem.Equals(item)) 
     { 
      foundTheItem = true; 
      break; 
     } 
    } 
    if (foundTheItem == false) 
    { 
     // Add here 
    } 
} 

Answer 2

이것은 흥미로운 질문입니다. 중복을 무시하는 차단 대기열을 요청한 사람은 이번이 처음입니다. 이상하게도 BCL에 이미있는 것과 같은 것을 찾을 수는 없습니다. BlockingCollection은 중복이 감지되었을 때 실패 할 수 있다고 광고되는 TryAdd 메서드를 가진 기본 컬렉션으로 IProducerConsumerCollection을 받아 들일 수 있기 때문에 이것이 이상하다고 말합니다. 문제는 중복을 방지하는 IProducerConsumerCollection의 구체적인 구현이 없다는 것입니다. 적어도 우리는 우리 자신을 쓸 수 있습니다.

public class NoDuplicatesConcurrentQueue<T> : IProducerConsumerCollection<T> 
{ 
    // TODO: You will need to fully implement IProducerConsumerCollection. 

    private Queue<T> queue = new Queue<T>(); 

    public bool TryAdd(T item) 
    { 
    lock (queue) 
    { 
     if (!queue.Contains(item)) 
     { 
     queue.Enqueue(item); 
     return true; 
     } 
     return false; 
    } 
    } 

    public bool TryTake(out T item) 
    { 
    lock (queue) 
    { 
     item = null; 
     if (queue.Count > 0) 
     { 
     item = queue.Dequeue(); 
     } 
     return item != null; 
    } 
    } 
} 

이제 우리는 중복을 허용하지 않는 우리의 IProducerConsumerCollection을 가지고 우리는 다음과 같이 사용할 수 있습니다 :

public class Example 
{ 
    private BlockingCollection<object> queue = new BlockingCollection<object>(new NoDuplicatesConcurrentQueue<object>()); 

    public Example() 
    { 
    new Thread(Consume).Start(); 
    } 

    public void Produce(object item) 
    { 
    bool unique = queue.TryAdd(item); 
    } 

    private void Consume() 
    { 
    while (true) 
    { 
     object item = queue.Take(); 
    } 
    } 
} 

당신은 NoDuplicatesConcurrentQueue 내 구현을 좋아하지 않을 수 있습니다. 당신은 확실히 ConcurrentQueue 또는 TPL 콜렉션이 제공하는 낮은 락 성능이 필요하다고 생각한다면 당신 자신을 구현할 수 있습니다.

업데이트 :

가 오늘 아침 코드를 테스트 할 수 있었다. 좋은 소식과 나쁜 소식이 있습니다. 좋은 소식은 이것이 기술적으로 효과가 있다는 것입니다. 나쁜 소식은 BlockingCollection.TryAdd 메서드가 기본 IProducerConsumerCollection.TryAdd 메서드에서 반환 값을 가로 채고 false이 검색되면 예외를 throw하기 때문에이 작업을 수행하지 않으려는 것입니다. 네, 맞습니다. 기대했던대로 false을 반환하지 않고 대신 예외를 생성합니다. 솔직히 말해서, 이것은 놀랍고 우습다. TryXXX 메서드의 요점은 예외를 던져서는 안된다는 것입니다. 나는 깊이 실망했다.브라이언 기드온이 업데이트후 언급 한주의뿐만 아니라

Answer 3

이 솔루션은 이러한 성능 문제를 앓고 :

• O 작업 큐 (queue.Contains(item)) 성능에 심각한 영향을 미칠 수에 (N) 큐가 성장함에 따라
• 잠금

다음 코드는 브리에 향상 (그가 언급 않음) 동시성을 제한 해시 세트를 사용

• 의하여의 용액을 수행 할 O (1) System.Collections.Concurrent 공간으로부터 2 개 데이터 구조를 조합 조회

N.B. ConcurrentHashSet이 없으므로 값을 무시하고 ConcurrentDictionary을 사용하고 있습니다.

이 드문 경우로, 잠금을 추가하지 않고 여러 개의 간단한 데이터 세트 중에서 더 복잡한 동시 데이터 구조를 간단히 작성할 수 있습니다. 여기서는 2 개의 동시 데이터 구조에 대한 연산 순서가 중요합니다.

public class NoDuplicatesConcurrentQueue<T> : IProducerConsumerCollection<T> 
{ 
    private readonly ConcurrentDictionary<T, bool> existingElements = new ConcurrentDictionary<T, bool>(); 
    private readonly ConcurrentQueue<T> queue = new ConcurrentQueue<T>(); 

    public bool TryAdd(T item) 
    { 
     if (existingElements.TryAdd(item, false)) 
     { 
      queue.Enqueue(item); 
      return true; 
     } 
     return false; 
    } 

    public bool TryTake(out T item) 
    { 
     if (queue.TryDequeue(out item)) 
     { 
      bool _; 
      existingElements.TryRemove(item, out _); 
      return true; 
     } 
     return false; 
    } 
    ... 
} 

N.B. 이 문제를 보는 또 다른 방법은 다음과 같습니다. 은 삽입 주문을 보존하는 세트입니다.