2

これは、宿題の生産者/消費者パターンの実装です。以下の実装の何が問題になっていますか。私はさまざまな実装をグーグルで検索しましたが、私の何がうまくいかなかったのか理解できません。

共有キューがあります

同じロックでプロデューサーとコンシューマーを同期します

実装

共有キュー：

 class SharedQueue{
    public static Queue<Integer>   queue  = new LinkedList<Integer>();
 }

プロデューサースレッド：

//The producer thread
class Producer implements Runnable{
    public void run()
    {
        synchronized (SharedQueue.queue)
        {
            if(SharedQueue.queue.size() >=5)
            {
                try {
                    SharedQueue.queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            Random r  = new Random();

            int x = r.nextInt(10);
            System.out.println("Inside Producer" + x);

            SharedQueue.queue.offer(x);


            SharedQueue.queue.notify();

        }
    }
}

コンシューマースレッド：

class Consumer implements  Runnable{
    public void run()
    {
        synchronized (SharedQueue.queue)
        {
            if(SharedQueue.queue.size() == 0)
            {
                try {
                    SharedQueue.queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
                }
            }

            int k = SharedQueue.queue.remove();

            System.out.println("Inside consumer" + k);
        }
    }
}

メインプログラム

public class ProducerConsumerTest {

    public static void main(String[] args)
    {
        
        Thread p = new Thread(new Producer());
        Thread q = new Thread(new Consumer());

        p.start();
        q.start();

    }
}
4

6 に答える 6

4

交換してみてください：

if(SharedQueue.queue.size() >= 5)

と：

while(SharedQueue.queue.size() >= 5)

この：

if(SharedQueue.queue.size() == 0)

と：

while(SharedQueue.queue.size() == 0)

を呼び出した後、条件を再確認するだけnotify()です。

于 2012-04-23T02:54:54.433 に答える
1

これをプロデューサーコンシューマーの無限のループにしたいと思います。Eng.Fouadの変更に加えて、同期された両方のブロックを次のように囲みます。

        while (true)

コンシューマーに通知を追加します

        int k = SharedQueue.queue.remove(); 

        // make the producer active again
        SharedQueue.queue.notify();

        System.out.println("Inside consumer " + k);
于 2012-04-23T03:49:10.610 に答える
1

生産者/消費者問題を実装する簡単な方法は、セマフォを使用することです。

public class Semaphore {
    int value;

    public Semaphore(int intialValue) {
        this.value = intialValue;
    }

    public synchronized void p() {
        while (value <= 0) {
            try {
                this.wait();
            } catch (InterruptedException e) {
            }
        }
        value = value - 1;
    }

    public synchronized void v() {
        value = value + 1;
        this.notify();
    }
}

public class ProducerConsumerUsingSemaphore {

    private static final int SIZE = 10;

    public static void main(String[] args) {

        Semaphore full = new Semaphore(0);
        Semaphore empty = new Semaphore(SIZE);
        Semaphore mutex = new Semaphore(1);
        Vector<Integer> sQueue = new Vector<Integer>();

        Thread producerThread = new Thread(new Runnable() {

            @Override
            public void run() {

                for (int i = 0; i < 5000; i++) {
                    empty.p();
                    mutex.p();
                    System.out.println(Thread.currentThread().getName() + " is trying to insert item " + i);
                    sQueue.add(i);
                    mutex.v();
                    full.v();
                }
            }
        });

        Thread consumerThread = new Thread(new Runnable() {

            @Override
            public void run() {
                while (true) {
                    full.p();
                    mutex.p();
                    System.out.println(Thread.currentThread().getName() + " consuming item " + sQueue.remove(0));
                    mutex.v();
                    empty.v();
                }
            }
        });

        producerThread.setName("Producer");
        consumerThread.setName("Consumer");

        consumerThread.start();
        producerThread.start();

    }
}
于 2015-12-16T03:59:47.787 に答える
0

ConcurrentLinkedQueueを使用して、プロデューサーとコンシューマーの共有キューを管理できます。ConcurrentHashMap>コレクションを利用できます。これにより、プロデューサーは同時に生成でき、コンシューマーは同時に消費し、プロデューサーによって生成されたキーを別のコレクションオブジェクトに保持できます。ここで、コンシューマーはキーを見つけてConcurrentHashMap>から消費できます。

于 2016-09-29T13:17:31.957 に答える
0 
public class ProducerAndConsumer {
    public static void main(String a[]) {
        Resource resource = new Resource();
        Producer producer = new Producer(resource);
        Consumer consumer = new Consumer(resource);
        producer.start();
        consumer.start();

    }
}

class Resource {
    private int item = 0;
    boolean flag = true;

    public void getItem() {
        while (true) {
            synchronized (this) {
                if (!flag) {
                    try {
                        System.out.println("Consumer consume item :" + item);
                        flag = true;
                        Thread.sleep(10);
                        notify();
                        wait();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        }

    }

    public void setItem() {
        while (true) {
            synchronized (this) {

                if (flag) {
                    try {
                        item++;
                        System.out.println("Producer creating item :" + item);
                        flag = false;
                        Thread.sleep(10);
                        notify();
                        wait();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }

            }
        }
    }

}

class Producer extends Thread {
    Resource resource = null;

    Producer(Resource resource) {
        this.resource = resource;
    }

    @Override
    public void run() {
        resource.setItem();
    }
}

class Consumer extends Thread {
    Resource resource = null;

    Consumer(Resource resource) {
        this.resource = resource;
    }

    @Override
    public void run() {
        resource.getItem();
    }
}
于 2019-06-16T08:25:23.843 に答える
0

毒薬で私のパターンを使用するだけです：

public sealed interface BaseMessage {

    final class ValidMessage<T> implements BaseMessage {

        @Nonnull
        private final T value;


        public ValidMessage(@Nonnull T value) {
            this.value = value;
        }

        @Nonnull
        public T getValue() {
            return value;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            ValidMessage<?> that = (ValidMessage<?>) o;
            return value.equals(that.value);
        }

        @Override
        public int hashCode() {
            return Objects.hash(value);
        }

        @Override
        public String toString() {
            return "ValidMessage{value=%s}".formatted(value);
        }
    }

    final class PoisonedMessage implements BaseMessage {

        public static final PoisonedMessage INSTANCE = new PoisonedMessage();


        private PoisonedMessage() {
        }

        @Override
        public String toString() {
            return "PoisonedMessage{}";
        }
    }
}

public class Producer implements Callable<Void> {

    @Nonnull
    private final BlockingQueue<BaseMessage> messages;

    Producer(@Nonnull BlockingQueue<BaseMessage> messages) {
        this.messages = messages;
    }

    @Override
    public Void call() throws Exception {
        messages.put(new BaseMessage.ValidMessage<>(1));
        messages.put(new BaseMessage.ValidMessage<>(2));
        messages.put(new BaseMessage.ValidMessage<>(3));
        messages.put(BaseMessage.PoisonedMessage.INSTANCE);
        return null;
    }
}

public class Consumer implements Callable<Void> {

    @Nonnull
    private final BlockingQueue<BaseMessage> messages;

    private final int maxPoisons;


    public Consumer(@Nonnull BlockingQueue<BaseMessage> messages, int maxPoisons) {
        this.messages = messages;
        this.maxPoisons = maxPoisons;
    }

    @Override
    public Void call() throws Exception {
        int poisonsReceived = 0;
        while (poisonsReceived < maxPoisons && !Thread.currentThread().isInterrupted()) {
            BaseMessage message = messages.take();
            if (message instanceof BaseMessage.ValidMessage<?> vm) {
                Integer value = (Integer) vm.getValue();
                System.out.println(value);
            } else if (message instanceof BaseMessage.PoisonedMessage) {
                ++poisonsReceived;
            } else {
                throw new IllegalArgumentException("Invalid BaseMessage type: " + message);
            }
        }
        return null;
    }
}
于 2020-10-03T05:50:00.580 に答える