java eşzamanlı: çok üretici bir tüketici

Question

Farklı iş parçacıklarının bir kuyruğu (üretici) ve bir tüketiciyi bu sıradan almasını sağlayan bir durum var. Benim sorunum, bu elemanlardan biri sıradan alındığında bazı cevapsız (eksik sinyal?).

class Producer implements Runnable { 

    private Consumer consumer; 

    Producer(Consumer consumer) { this.consumer = consumer; } 

    @Override 
public void run() { 
    consumer.send("message"); 
    } 
} 

ve oluşturdukları ve birlikte çalıştırılır: üreticileri kodudur

ExecutorService executor = Executors.newSingleThreadExecutor(); 
for (int i = 0; i < 20; i++) { 
    executor.execute(new Producer(consumer)); 
} 

Tüketici kodudur:

class Consumer implements Runnable { 

private Queue<String> queue = new ConcurrentLinkedQueue<String>(); 

void send(String message) { 
    synchronized (queue) { 
     queue.add(message); 
     System.out.println("SIZE: " + queue.size()); 
     queue.notify(); 
    } 
} 

@Override 
public void run() { 
    int counter = 0; 
    synchronized (queue) { 
    while(true) { 
     try { 
      System.out.println("SLEEP"); 
       queue.wait(10); 
     } catch (InterruptedException e) { 
       Thread.interrupted(); 
     } 
     System.out.println(counter); 
     if (!queue.isEmpty()) {    
      queue.poll(); 
      counter++; 
     } 
    } 
    } 
} 

} 

kod çalıştırıldığında bazen 20 element eklendi olsun ve 20 alındı, ancak diğer durumlarda alınan öğeler 20'den küçüktür. Herhangi bir fikrin nasıl düzeltileceği.

Answer 1

Bir Kuyruk yerine bir BlockingQueue kullanmanız önerilir. Bir LinkedBlockingDeque sizin için iyi bir aday olabilir.) (Edecektir

void send(String message) { 
    synchronized (queue) { 
     queue.put(message); 
     System.out.println("SIZE: " + queue.size()); 
    } 
} 

ve sonra fikir sökünüz olan tüketici iplik

üzerinde sadece

queue.take() 

gerekir:

Kodunuz şu şekilde görünecektir Kuyrukta bir öğe bulunana kadar bloklayın ve ardından tam olarak bir (uygulamada zarara uğradığımı düşünüyorum: eksik bildirim e oylama). .put() sizin için tüm bildirimleri yapmaktan sorumludur. Beklemede/bildirim gerektirmez.

Answer 2

Kodunuzdaki sorun, notifyAll yerine notify kullanıyor olmanızdır. Birincisi, kilitte bekleyen biri varsa, sadece tek bir iplik uyandırır. Bu, hiçbir ipliğin beklemediği ve sinyalin kaybolduğu bir yarış koşuluna izin verir. Bir bildirim, tüm iş parçacıklarının kilidi elde edip edemeyeceklerini kontrol etmek için uyanmalarını gerektirerek, küçük bir performans maliyetinde doğruluğu zorlayacaktır.

Bu en iyi Effective Java 1st ed numaralı belgede açıklanmıştır (bkz. S.150). Programcılar, daha güçlü doğruluk garantileri sağlayan java.util.concurrent'ı kullanmaları beklendiğinden, 2. baskı bu ucu çıkardı.

Answer 3

ConcurrentLinkedQueue ve eşitlemeyi aynı anda kullanmak kötü bir fikir gibi görünüyor. Eşzamanlı veri yapılarının amacını ilk etapta ortadan kaldırır.

ConcurrentLinkedQueue veri yapısında sorun yok ve BlockingQueue ile değiştirilirse sorun çözülür, ancak bu kök neden değildir.

Sorun queue.wait (10) ile ilgilidir. Bu zamanlanmış bekleme yöntemidir. 10 ms sonra tekrar kilitlenecektir.

1. Bildirim 10ms geçmişse, üzerinde bekleyen hiçbir tüketici iplik olmadığı için (queue.notify()) kaybolmak. Kilitleme, tüketici tarafından tekrar talep edildiğinden, kilitlenemediği için, üretici kuyruğa eklenemeyecektir.Eğer bekleme (10) kod kaldırıldı ve bekleyin ve BlockingQueue veri yapısıyla dikkat çekildiği bildirmek çünkü BlockingQueue taşıma

sorununuzu çözdü.