Productor/Consumidor en Java

[moisesvs]

Hola a todos!

He intentado generar el código para hacer el programa de productor/consumidor, pero no se por que no entiendo muy los synchronized en Java y no me funciona correctamente, el código que he generado es el siguiente haber si me podeís decir algo de como solucionar para que no se bloqueen o alguna idea. Muchas gracias!

Código:

import java.util.*;

public class Intercambio extends Thread{
        //0 Consumidor
        //1 Productor
        private int tipo;
        private static LinkedList <Proceso> permisosponer = new LinkedList();
        private static LinkedList <Proceso>permisosquitar = new LinkedList();
       
        public Intercambio(int ptipo)
        {
            this.tipo=ptipo;
        }
       
        public void run ()
        {
            if(this.tipo==0)
                for(;;)
                    get();
            else
                for(;;)
                    put();
        }
       
        public void get()
        {
            try{
                Proceso p = new Proceso();
                synchronized (p)
                {
                    synchronized (permisosponer)
                    {
                      if (permisosponer.size()==0)
                      {
                               System.out.println("CONSUMIDOR ME DUERMO...");
                             permisosponer.wait();
                      }
                          permisosquitar.add(p);
                          System.out.println("CONSUMIDOR ME CONSUMIENDO!!");
                          permisosquitar.notify();
                    }
                }
            }catch (InterruptedException e)
            {
                System.out.println("PROBLEMA AL PARAR EL HILO");
            }
        }
        public void put()
        {
            try{
                Proceso p = new Proceso();
                synchronized (p)
                {
                    synchronized (permisosquitar)
                    {
                      if (permisosquitar.size()==0)
                      {
                              System.out.println("\t\tPRODUCTOR ME DUERMO...");
                             permisosquitar.wait();
                      }
                      permisosponer.add(p);
                      System.out.println("\t\tPRODUCTOR ESCRIENDO!");
                      permisosponer.notify();
                    }
                }
            }catch (InterruptedException e)
            {
                System.out.println("PROBLEMA AL PARAR EL HILO");
            }
        }
}

[moisesvs]

Nadie me puede ayudar...

Gracias!

[moisesvs]

Creo que lo he conseguido!

Código:

import java.util.*;

public class Intercambio extends Thread{
		//0 Consumidor
		//1 Productor
		private int tipo;
		private static Proceso permisosponer = new Proceso();
		private static Proceso permisosquitar = new Proceso();
		private static boolean producido = false;
		private static int num_producido = 0;
		
		
		public Intercambio(int ptipo)
		{
			this.tipo=ptipo;
			
			
		}
		
		public void run ()
		{
			if(this.tipo==0)
				for(;;)
					get();
			else
				for(;;)
					put();
		}
		
		public void get()
		{
			try{
				synchronized (permisosquitar)
				{
					  if (!(this.producido))
					  {
						  	 System.out.println("CONSUMIDOR ME DUERMO...");
						     permisosquitar.wait();
					  }
				}
				synchronized (permisosponer)
				{
	
						System.out.println("CONSUMIDOR CONSUMIENDO!!"+"\t"+this.num_producido);
						
						this.producido=false;
						permisosponer.notify();
				}

			}catch (InterruptedException e)
			{
				System.out.println("PROBLEMA AL PARAR EL HILO");
			}
		}
		public void put()
		{
			try{
					synchronized (permisosponer)
					{
					  if (this.producido)
					  {
						  	System.out.println("\t\tPRODUCTOR ME DUERMO...");
						    permisosponer.wait();
					  }
					}
					synchronized (permisosquitar)
					{
					  System.out.println("\t\tPRODUCTOR ESCRIENDO!");
					  this.num_producido=(int)Math.random();
					  this.producido=true;
					  permisosquitar.notify();
					}
			}catch (InterruptedException e)
			{
				System.out.println("PROBLEMA AL PARAR EL HILO");
			}
		}
}

[megajandro]

me parece que no tiene mucho sentido el codigo que mostras. Si queres verlo un poco mas facil pensalo en lugar de como productor/consumidor como cliente/servidor. Donde el cliente es el producto y el servidor el consumidor. La idea es que no uses el sincronized, sino que generes un thread por cada requerimiento y lo dejes encolado, y cuando alguno de tus thread consumidores quede disponible lo consuma los que podemos llamar listener. Buscate como ejemplo para practicarlo en spring-remoting con AMQ tenes muchos de estos ejemplos.

[moisesvs]

No se lo que me quieres decir, pero la idea es que el código esta mal... por que pude que entren dos consumidores a la vez.

Código:

 public void get()
{
try{
synchronized (permisosquitar)
{
 if (!(this.producido))
 {
  System.out.println("CONSUMIDOR "+this.num+" ME DUERMO...");
    permisosquitar.wait();
 }
}
----------------------------------------------------------------
synchronized (permisosponer)
{
System.out.println("CONSUMIDOR "+this.num+" CONSUMIENDO!!"+"\t"+this.num_producido);
this.producido=false;
permisosponer.notify();
}

}catch (InterruptedException e)
{
System.out.println("PROBLEMA AL PARAR EL HILO");
}
}

En la linea esa se pueden meter dos consumidores a la vez, con lo cual, pueden consumir el mismo numero a la vez.

Como podría solucionar este hecho.

Saludos y gracias!

[megajandro]

fijate si te sirve este ejemplo.


public class WorkQueue {
private final int nThreads;
private final PoolWorker[] threads;
private final LinkedList queue;

public WorkQueue(int nThreads) {
this.nThreads = nThreads;
queue = new LinkedList();
threads = new PoolWorker[nThreads];
for (int i = 0; i < nThreads; i++) {
threads[i] = new PoolWorker();
threads[i].start();
}
}

public void execute(Runnable r) {
synchronized (queue) {
queue.addLast(r);
queue.notify();
}
}

private class PoolWorker extends Thread {
public void run() {
Runnable r;
while (true) {
synchronized (queue) {
while (queue.isEmpty()) {
try {
System.out.println("Estoy esperando!!");
queue.wait();
} catch (InterruptedException ignored) {
}
}
r = (Runnable) queue.removeFirst();
System.out.println("fue consumido: " + r);
}
// If we don't catch RuntimeException,
// the pool could leak threads
try {
r.run();
} catch (RuntimeException e) {
// You might want to log something here
}
}
}
}
}

public class Productor implements Runnable{

public void run() {
System.out.println("estoy ejecutando: " + this);
}
}


public class TestWorkQueue extends TestCase{

public void test(){
WorkQueue workQueue = new WorkQueue(3);
for(int i=0; i < 10;i++){
Runnable r = new Productor();
workQueue.execute(r );

}
}
}