Здравствуйте.

имеется работающая реализация dataQueue с Producer и Receiver

Поток продьюсера считаывает горизонтально дюжину файлов, сортирует содержимое,
приводит в один List<DataItem> и обновляет цепочку данных

Поток ресивера считывает без промедления цепочку данных.

Потоки продьюсера и ресивера синхронизированы на сигнальный обьект, следящий за размером цепочки и концами файлов.

В ресивере каждый полученный DataItem должен будить потоки группы Наблюдателей
и без модификции считываться в них параллельно.

Пробная реализация Runnable Обсервера, с попыткой пробуждения, усыпления потока
через volatile поля не работает, входные данные (Integer) не обрабатывается.

Мне будут интересны указания на мои ошибки с реализацией

void run(){
  synchronized(this){
  }
}

с уважением,
обьект

////////////////////////////////////////////////////////
public class ObserverTest {

    static class Observer implements Runnable{
       private volatile boolean threadSuspended;
       private volatile boolean threadFinished;
       private Integer item;

        Observer(){
            threadSuspended = true;
            threadFinished = false;
        }

        public boolean isThreadSuspended() {
            return threadSuspended;
        }

        public Integer getItem() {
            return item;
        }
        public void setItem(Integer item) {
            this.item = item;
            setThreadSuspended(false); //
        }

        public void setThreadSuspended(boolean threadSuspended) {
            this.threadSuspended = threadSuspended;
        }

        public void setThreadFinished() {
            this.threadFinished = true;
        }

        public void run(){

            long interval = 10;  //in ms
            Thread thisThread = Thread.currentThread();
            synchronized(this){
                while (true) {
                    try {
                        thisThread.sleep(interval);


                        while (threadSuspended){
                            wait();
                        }

                        if (threadFinished)
                            break;

                    //processing
                        System.out.println(thisThread.getId()+" id  processing item: "+item);

                    //putting thread back to wait  state
                        this.setThreadSuspended(true);

                    } catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    public static void main(String[] args) {
        Observer  obs = new Observer();
        Thread observerT = new Thread(obs);
        
        observerT.start();

        obs.setItem(new Integer(1));
        obs.setItem(new Integer(2));
        obs.setItem(new Integer(3));

        obs.setThreadFinished();
    }
}