1

我有一个标准的生产者消费者问题。生产者将数据放入堆栈（缓冲区）消费者接受它。问题是当生产者结束生产数据时，消费者并没有死亡（并非总是如此）。

我想让消费者在生产者结束他的for循环后死去

for(int i = 0; i < 10; i++){
        try{
    //      sleep((int)(Math.random() * 1));                
        }catch(Exception e){e.printStackTrace();}
        b.put((int) (Math.random()* 10));
        System.out.println("i = " + i);
    }
    b.stop();

那么我将b.stop()which changes runningfield in称为Bufferfalse 和notifiesAll()

结束然后我得到：

i = 9 // number of iteration this is 10th iteration
Consumer 2.: no data to take. I wait.  Memory: 0
Consumer 1.: no data to take. I wait.  Memory: 0
Consumer 3.: no data to take. I wait.  Memory: 0

他们应该死了，所以我做了方法 stop() 但它没有用。

代码正在运行，请检查

import java.util.Stack;


public class Buffer {
private static int SIZE = 4;
private int i;//number of elements in buffer
public Stack<Integer> stack;
private volatile boolean running;
    public Buffer() {
        stack = new Stack<>();
        running = true;
        i = 0;
    }
    synchronized public void put(int val){
        while (i >= SIZE) {
            try {
                System.out.println("Buffer full, producer waits");
                wait();
            } catch (InterruptedException exc) {
                exc.printStackTrace();
            }
        }   
        stack.push(val);//txt = s;
        i++;
        System.out.println("Producer inserted " + val + " memory: " + i);
        if(i - 1 == 0)
            notifyAll();
        System.out.println(stack);
    }

    public synchronized Integer get(Consumer c) {
        while (i == 0) {
            try {
                System.out.println(c + ": no data to take. I wait.  Memory: " + i);
                wait();
            } catch (InterruptedException exc) {
                exc.printStackTrace();
            }
        }   
        if(running){
            int data = stack.pop();
            i--;    
            System.out.println(c+  ": I took: " + data +" memory: " +  i);
            System.out.println(stack);
            if(i + 1 == SIZE){//if the buffer was full so the producer is waiting
                notifyAll();
                System.out.println(c +  "I notified producer about it");
        }
        return data;}
        else 
            return null;
    }

    public boolean isEmpty(){
        return i == 0;
    }
    public synchronized void stop(){//I THOUGH THIS WOULD FIX IT~!!!!!!!!!!!!!!
        running = false;
        notifyAll();
    }
    public boolean isRunning(){
        return running;
    }

}

public class Producer extends Thread {
private Buffer b;
    public Producer(Buffer b) {
        this.b = b;
    }

    public void run(){
        for(int i = 0; i < 10; i++){
            try{
        //      sleep((int)(Math.random() * 1));                
            }catch(Exception e){e.printStackTrace();}
            b.put((int) (Math.random()* 10));
            System.out.println("i = " + i);
        }
        b.stop();
    }

}

public class Consumer extends Thread {
    Buffer b;
    int nr;
    static int NR = 0;

    public Consumer(Buffer b) {
        this.b = b;
        nr = ++NR;
    }

    public void run() {
        Integer i = b.get(this);
        while (i != null) {
            System.out.println(nr + " I received : " + i);
            i = b.get(this);
        }
        System.out.println("Consumer " + nr + " is dead");
    }

    public String toString() {
        return "Consumer " + nr + ".";
}

}

public class Main {

    public static void main(String[] args) {

        Buffer b = new Buffer();
        Producer p = new Producer(b);
        Consumer c1 = new Consumer(b);
        Consumer c2 = new Consumer(b);
        Consumer c3 = new Consumer(b);  
        p.start();
        c1.start();c2.start();c3.start();

    }

}
4

4 回答 4

2

当缓冲区为空时，消费者开始等待。当收到通知时，它会检查缓冲区是否为空，如果仍然为空，则再次开始等待。running如果标志已设置为 false ，则不应再次开始等待：

while (i == 0 && running) {
    ...
于 2013-04-20T13:02:07.330 回答
0

run()您的for中的一个小修改Consumer：

public void run() {
    Integer i = b.get(this);
    while (i != null && i!=10) {
        System.out.println(nr + " I received : " + i);
        i = b.get(this);
    }
    System.out.println("Consumer " + nr + " is dead");
}
于 2013-04-20T13:03:11.127 回答
0

您的实现存在重大缺陷。变量 i 的修改不是线程安全的，这意味着您可以获得无法解释的结果。它可能导致竞争状况。

于 2013-04-20T13:08:08.817 回答
0 
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace ThreadsConsolApp
{

        public sealed class ProducerConsumer
        {
            const int MagicNumber = 30;                                 // Indicates how many times to bounce between ping and pong threads 
            private Object m_lock = new Object();                       // Lock to protect counter increment 
            private Queue<int> m_queue = new Queue<int>();

            // Ctor 
            public ProducerConsumer()
            {
            }
            public void admin()
            {
                int i = 0;
                lock (m_lock)
                {
                    while (i <= MagicNumber)
                    {
                        Monitor.Wait(m_lock); //wait
                        Console.WriteLine("Controller = Produced " + i + " , Consumeed " + i);
                        Monitor.Pulse(m_lock); //release
                        i++;
                    }
                }
            }
            // Ping
            public void Producer()
            {
                int counter = 0;

                lock (m_lock)                                               // Allows only one thread at a time inside m_lock 
                {
                    while (counter <= MagicNumber)
                    {
                        Thread.Sleep(500);                               // Get data chunks from some source 
                        Monitor.Wait(m_lock);                           // Wait if the thread is busy. 'wait' will hold 
                                                                       //this loop until something else pulses it to release the wait. 
                        Console.WriteLine("producer {0}", counter);
                        m_queue.Enqueue(counter);//entring in queue
                        Monitor.Pulse(m_lock);                          // Releases consumer thread 
                        counter++;
                    }
                }
            }

            public void Consumer()
            {
                lock (m_lock)                                           // Allows only one thread at a time inside m_lock 
                {
                    Monitor.Pulse(m_lock);

                    while (Monitor.Wait(m_lock,1000))                  // Wait in the loop while producer is busy. Exit when producer times-out. 1000 = 1 second; ...
                                                                      //app will hang without this time-out value 
                    {
                        int data = m_queue.Dequeue();//dispatch from queue
                        Console.WriteLine("consumer {0}", data);
                        Monitor.Pulse(m_lock);                          // Release consumer 
                        Console.WriteLine("=====================");
                    }
                }
            }
        }

        class Program
        {
            static void Main(string[] args)
            {
                ProducerConsumer app = new ProducerConsumer();

                // Create 2 threads 
                Thread t_producer = new Thread(new ThreadStart(app.Producer));
                Thread t_consumer = new Thread(new ThreadStart(app.Consumer));
                Thread t_admin = new Thread(new ThreadStart(app.admin));

                // Start threads 
                t_admin.Start();
                t_producer.Start();
                t_consumer.Start();

                // Waith for the threads to complete 
                t_admin.Join();
                t_producer.Join();
                t_consumer.Join();

                Console.WriteLine("\nPress any key to complete the program.\n");
                Console.ReadKey(false);
            }
        }
    }
于 2014-01-09T10:23:13.067 回答