Производитель Потребитель в Java многопоточный
Я пытаюсь реализовать проблему потребителя производителя в Java. Я использую кольцевой буфер (круговой массив), чтобы продюсер вставлял элементы в буфер. Ниже приведен мой код:
import java.util.*;
import java.io.*;
public class Buffer
{
String a[];
int front, rear;
public Buffer(int size)
{
a = new String[size];
front = rear = -1;
}
public boolean insert(String dataitem)
{
int p;
p = (rear+1) % a.length;
if(p==front)
{
System.out.println("Buffer full");
return false;
}
else
{ rear = p;
a[rear] = dataitem;
if(front == -1)
front = 0;
return true;
}
}
public boolean empty()
{
if(front == -1)
return true;
else
return false;
}
public String delete()
{
String result = a[front];
if(front == rear)
front = rear = -1;
else
front = (front +1)%a.length;
return result;
}
public void display()
{
if(front == -1)
System.out.println("Buffer empty");
else
{
System.out.println("Buffer elements are:");
int i= front;
while(i!= rear)
{
System.out.println(a[i]);
i = (i+1)%a.length;
}
System.out.println(a[i]);
}
}
public static void main(String[] args)
{
int size = Integer.parseInt(args[0]);
Buffer b = new Buffer(size);
int ch;
String dataitem, msg;
Thread prod = new Thread(new Producer(b, size));
Thread cons = new Thread(new Consumer(b, size));
prod.start();
cons.start();
}
}
class Producer extends Thread
{
Buffer b;
int size;
public Producer(Buffer b, int size)
{
this.b = b;
this.size = size;
}
public void run()
{
while(true)
{
synchronized(b)
{
for(int i = 1; i <= size; i++)
{
try
{ String dataitem = Thread.currentThread().getId()+"_"+i;
boolean bool = b.insert(dataitem);
//b.notifyAll();
if(bool)
System.out.println("Successfully inserted "+dataitem);
b.notifyAll();
Thread.sleep(2000);
}
catch(Exception e)
{ e.printStackTrace();
}
}
}
}
}
}
class Consumer extends Thread
{
Buffer b;
int size;
public Consumer(Buffer b, int size)
{
this.b = b;
this.size = size;
}
public void run()
{
while(b.empty())
{
synchronized(b)
{
try
{
System.out.println("Buffer empty");
b.wait();
}
catch(Exception e)
{ e.printStackTrace();
}
}
}
synchronized(b)
{
b.notifyAll();
String dataitem = b.delete();
System.out.println("Removed "+dataitem);
}
}
}
Производитель успешно вставляет элементы данных в буфер. Но они не потребляются потребителем.
Я получаю следующий вывод при выполнении программы.
Successfully inserted 11_1
Successfully inserted 11_2
Buffer full
Buffer full
Buffer full
Buffer full
Buffer full
Buffer full
Мой вопрос: как заставить потребителя потреблять элементы из буфера?
1 ответ
Основная проблема заключается в том, что synchronized блок в вашем Producer слишком широк Это никогда не позволяя Consumer получить замок
Начните с сужения области, например...
while (true) {
for (int i = 1; i <= size; i++) {
try {
String dataitem = Thread.currentThread().getId() + "_" + i;
boolean bool = b.insert(dataitem);
//b.notifyAll();
if (bool) {
System.out.println("Successfully inserted " + dataitem);
}
synchronized (b) {
b.notifyAll();
}
Thread.sleep(2000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
Вы также можете рассмотреть возможность синхронизации insert а также delete сами методы. Лично я хотел бы использовать внутреннюю блокировку, но вы могли бы просто synchronize сами методы, например...
public synchronized boolean insert(String dataitem) {...}
public synchronized String delete() {...}
Как оно есть, ваш Consumer будет когда-либо читать только одно значение из буфера, но я дам вам понять это;)
Как примечание, я мог бы поместить функцию ожидания и уведомления непосредственно в Bufferтак что всякий раз, когда вы пытаетесь delete значение будет ждать в пределах delete метод для Buffer быть не пустым и позволить insert способ сделать уведомления сам... но это я;)
Точно так же я мог бы рассмотреть возможность блокировки insert метод, пока есть больше места, но это будет сводиться к тому, как вы хотите его реализовать:P
обновленный
По сути, это начнет давать результаты, которые вы ищете...
public class ProducerConsumer {
public static void main(String[] args) {
new ProducerConsumer();
}
public ProducerConsumer() {
int size = 5;
Buffer b = new Buffer(size);
Thread prod = new Thread(new Producer(b, size));
Thread cons = new Thread(new Consumer(b, size));
prod.start();
cons.start();
}
public class Buffer {
String a[];
int front, rear;
public Buffer(int size) {
a = new String[size];
front = rear = -1;
}
public synchronized boolean insert(String dataitem) {
int p;
p = (rear + 1) % a.length;
if (p == front) {
System.out.println("Buffer full");
return false;
} else {
rear = p;
a[rear] = dataitem;
if (front == -1) {
front = 0;
}
return true;
}
}
public boolean empty() {
return front == -1;
}
public synchronized String delete() {
String result = a[front];
if (front == rear) {
front = rear = -1;
} else {
front = (front + 1) % a.length;
}
return result;
}
public void display() {
if (front == -1) {
System.out.println("Buffer empty");
} else {
System.out.println("Buffer elements are:");
int i = front;
while (i != rear) {
System.out.println(a[i]);
i = (i + 1) % a.length;
}
System.out.println(a[i]);
}
}
}
class Producer extends Thread {
Buffer b;
int size;
public Producer(Buffer b, int size) {
this.b = b;
this.size = size;
}
public void run() {
int i = 0;
while (true) {
try {
String dataitem = Thread.currentThread().getId() + "_" + ++i;
boolean bool = b.insert(dataitem);
if (bool) {
System.out.println("Successfully inserted " + dataitem);
}
synchronized (b) {
b.notifyAll();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
class Consumer extends Thread {
Buffer b;
int size;
public Consumer(Buffer b, int size) {
this.b = b;
this.size = size;
}
public void run() {
while (true) {
while (b.empty()) {
synchronized (b) {
try {
System.out.println("Buffer empty");
b.wait();
} catch (Exception e) {
e.printStackTrace();
}
}
}
String dataitem = null;
synchronized (b) {
dataitem = b.delete();
}
System.out.println("Removed " + dataitem);
}
}
}
}