java - Java多线程程序中生产者消费者线程的互换角色

Question

目标:创建两个线程，使生产者和消费者线程可以互换工作，即如果第一个线程充当生产者，则第二个线程充当消费者，反之亦然。

详细信息:它们通过缓冲区相互通信，存储一个整数大小。例如，如果第一个线程生成 1，那么第二个线程消耗它并生成 2，然后第一个线程消耗 2 并生成接下来的三个整数，Consumer 一个接一个地消耗它们。之后两个线程都会终止。此外，两个线程都应该能够启动通信。

我尝试编写以下代码。

import java.util.Random;
class CommonItem {
boolean flag = false;
int arr[];

public synchronized void Send(String msg) {
    if (flag) {
        try {
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    System.out.println(msg);
    flag = true;
    notify();
}

public synchronized void Receive(String msg) {
    if (!flag) {
        try {
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    System.out.println(msg);
    arr = send_random();
    for (int item: arr) {
        System.out.println(item);
    }

    flag = false;
    notify();
}

synchronized int[] send_random(){
    int[] arr = new int[3];
    Random random= new Random();
    for (int i = 0; i < 3; i++) {
        arr[i]=random.nextInt(100);
    }
    return arr;
}
}
class T1 implements Runnable {
CommonItem Ci;

public T1(CommonItem Ci) {
    this.Ci = Ci;
    new Thread(this, "producer").start();
}

public void run() {
    while (true)
    Ci.Send("sent :1");
}
}

class T2 implements Runnable {
CommonItem Ci;

public T2(CommonItem m2) {
    this.Ci = m2;
    new Thread(this, "Consumer").start();
}

public void run() {
    while (true)
    Ci.Receive("received :2");
}
}
public class TestClass {

public static void main(String[] args) {

    CommonItem m = new CommonItem();
    new T1(m);
    new T2(m);
}
}

预期输出是

sent :1

received :1

sent :2

received :2

sent :57 4 13

received :57 4 13

但是我得到以下输出输出

sent :1

received :2

57

4

13

请建议代码中是否有任何更正或关于如何以替代方式解决给定问题的任何想法。预先感谢您。

Answer 1

public class CommonItem {
    boolean receiver = false;
    List<Integer> list = new ArrayList<>();

    public void receive() throws InterruptedException {
        String name = Thread.currentThread().getName();
        synchronized (list) {
            while (list.isEmpty()) {
                list.notify();
                list.wait();
            }

            // Receive all elements
            System.out.printf("Receiving elements by %s:\t", name);
            for (int val : list) {
                System.out.print(val + "  ");
            }
            list.clear();
            System.out.println();
            list.notify();
            list.wait();
        }
    }

    public void send() throws InterruptedException {
        String name = Thread.currentThread().getName();
        synchronized (list) {
            while (!list.isEmpty()) {
                list.notify();
                list.wait();
            }
            // Sending elements
            int[] arr = get_random();
            System.out.printf("Sending elements by %s\t", name);
            for (int ele : arr) {
                list.add(ele);
                System.out.print(ele + "  ");
            }
            System.out.println();
            list.notify();
            list.wait();
        }
    }

    public int[] get_random() throws InterruptedException {
        int[] arr = new int[3];
        Random random = new Random();
        for (int i = 0; i < 3; i++) {
            arr[i] = random.nextInt(100);
        }
        Thread.sleep(1000);
        return arr;
    }
}

public class ThreadTask implements Runnable {

    private CommonItem item;
    private boolean receiver;

    public ThreadTask(CommonItem item, boolean receiver) {
        this.item = item;
        this.receiver = receiver;
    }

    public static void main(String[] args) {
        CommonItem item = new CommonItem();
        Thread t1 = new Thread(new ThreadTask(item, false), "First");
        Thread t2 = new Thread(new ThreadTask(item, true), "Second");
        t1.start();
        t2.start();
    }

    @Override
    public void run() {
        while (true) {
            try {
                if (receiver) {
                    item.receive();
                } else {
                    item.send();
                }
                receiver = !receiver;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

Sending elements by First 25 6 57
Receiving elements by Second: 25 6 57
Sending elements by Second 35 99 10
Receiving elements by First: 35 99 10
Sending elements by First 84 11 1
Receiving elements by Second: 84 11 1
Sending elements by Second 68 91 53
Receiving elements by First: 68 91 53