Что касается многопоточности JAVA, правильна ли эта программа подписки издателя?

Question

package com.mypublishersubscriber;

import java.util.LinkedList;
import java.util.Queue;
import java.util.Random;

class Publisher implements Runnable {

    Channel object = null;
    Random random = new Random();

    public Publisher(Channel object) {
        this.object = object;
    }

    public void run() {

        synchronized (object) {

            while (true) {
                if (object.queue.size() == object.capacity) {
                    try {
                        object.wait();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
                int ele  = random.nextInt();
                System.out.println("put: "+ ele);
                object.queue.add(ele);
                object.notify();
            }

        }

    }

}

class Consumer implements Runnable {

    Channel object = null;

    public Consumer(Channel object) {
        this.object = object;
    }

    public void run() {

        synchronized (object) {
            
            while(true) {
                
                if (object.queue.isEmpty()) {
                    try {
                        object.wait();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                } else {

                    int ele = object.queue.poll();
                    System.out.println("read: "+ele);
                    object.notify();

                }
                
            }

            

        }

    }

}
/**
 * publisher and subscriber will connect to this main class and<br>
 * will publish to this object's queue and consume from this
 * @author subhasish
 *
 */
public class Channel {

    Queue<Integer> queue = new LinkedList<>();
    public final int capacity = 1;

    public static void main(String[] args) {

        Channel object = new Channel();

        Publisher publisher = new Publisher(object);
        Consumer consumer = new Consumer(object);

        Thread producer = new Thread(publisher);
        Thread subscriber = new Thread(consumer);

        producer.start();
        subscriber.start();

    }

}

• Я создал класс Channel, «объект» которого используется как издателем, так и подписчиком.
• Издатель имеет синхронизированный блок в методе выполнения, в котором проверяется бесконечное l oop если очередь «канала» заполнена, если она заполнена, то если вызовы ожидают объекта «канал» и снимают блокировку, иначе добавляются элементы к нему и уведомляются, что в конечном итоге снова снимает блокировку
• Consumer также имеет синхронизированный блок в методе выполнения с бесконечным l oop, чтобы проверить, пуста ли очередь «канала», если да, вызывает ожидание на «канале» и снимает блокировку, иначе добавляет элемент в «канал» очередь и уведомляет.

СОМНЕНИЕ:

• Правильно ли я использую объект канала?
• Правильно ли я использую очередь «канала», учитывая ее совместно с двумя потоками, мне нужно использовать volatile или что-то в этом роде?

Программа работает, и я получаю правильный результат. Я знаю, что есть продвинутые способы сделать это, но я хочу понять, что такое ожидание и уведомить таким образом.

Answer 1

Вы правильно используете объект канала, потому что вы обращаетесь к нему только в синхронизированных блоках. С явной синхронизацией вам не нужно использовать volatile, поскольку операции синхронизации включают необходимый барьер памяти.

Однако ваш while l oop находится внутри синхронизированного блока, поэтому, когда один поток запускается, другой будет ждать, пока запущенный поток не вызовет wait, то есть пока очередь не заполнится или не опустеет. Вы можете подумать о перемещении синхронизированного блока в while l oop.