使用线程生产和消费到列表？

Question

我创建了两个单独的线程，一个用于写入列表，第二个用于从列表中删除。

package com.produ.consu;

public class Test {

    public static void main(String[] args) {

        Operations operations = new Operations();

        Runnable r1 = new ThreadsClass(operations);

        Runnable r2 = new ThreadsClass(operations);

        Thread t1 = new Thread(r1);

        Thread t2 = new Thread(r2);

        t1.setName("READ");

        t2.setName("WRITE");

        t1.start();

        t2.start();

    }

}

以上是我创建线程的测试类。

package com.produ.consu;

public class ThreadsClass implements Runnable {

    Operations operations;

    ThreadsClass(Operations operations){

        this.operations=operations;

    }

    @Override

    public void run() {

        // TODO Auto-generated method stub

        if(Thread.currentThread().getName().equals("WRITE")) {

            operations.writeList();

        }

        else {

            operations.readList();

        }

    }

}

以上是基于线程名称调用同步方法的类：

import java.util.ArrayList;

import java.util.List;

public class Operations {

    List<Integer> list=null;

    int count=0;    

    boolean flag;

    Operations(){

        list=new ArrayList<>();

        flag=true;

    }

    public  synchronized void writeList() {

        // TODO Auto-generated method stub

        while(true) {

            if(flag) {

                count++;

                list.add(count);

                System.out.println("inise if block...."+Thread.currentThread().getName());

                System.out.println(list);

                flag=false;

                try {

                    wait();

                } catch (InterruptedException e) {

                    // TODO Auto-generated catch block

                    e.printStackTrace();

                }

            }

            else {

                notify();

            }

        }

    }

上面是我提到功能的地方。 所以写线程必须将元素写入列表并等待它被第二个线程删除。一旦删除第二个应该通知第一个并等待直到元素插入。但是得到...

inise if block....写 [1]

甚至没有被删除，它必须是一个连续的过程。给我关于给定代码的建议。

Answer 1

您也应该wait()在else块内，否则while如果不满意，它将继续运行if并且其他线程将没有机会执行。执行读/写操作后，线程应调用notify以唤醒另一个线程，并且本身应进入等待状态。&nbsp;public synchronized void writeList() {&nbsp; &nbsp; &nbsp; while (true) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if (flag) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; count++;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; list.add(count);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; System.out.println("inise if block...." + Thread.currentThread().getName());&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; System.out.println(list);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; flag = false;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; notify(); //notify the read thread that write is complete&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; wait();&nbsp; &nbsp;// go into the waiting state so that no further write is done until the current element is removed by the read thread.&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } catch (InterruptedException e) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; e.printStackTrace();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } else {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; wait(); //wait in else, otherwise while will run endlessly&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } catch (InterruptedException e) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; e.printStackTrace();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;}&nbsp; &nbsp; &nbsp; }&nbsp;}&nbsp;public synchronized void readList() {&nbsp; &nbsp; &nbsp; &nbsp; while (true) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; System.out.println("in read");&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if (!flag) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Integer i = list.remove(0);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; System.out.println(i + "..removed at index by" + Thread.currentThread().getName());&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; flag = true;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; notify(); //notify write thread that read is complete&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; wait();&nbsp; &nbsp;//go into wait until new element is inserted&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } catch (InterruptedException e) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; e.printStackTrace();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } else {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; wait(); //wait in else otherwise while runs endlessly&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } catch (InterruptedException e) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; e.printStackTrace();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }

Answer 2

您应该用更细粒度的同步替换方法同步：使用同步块包装访问列表和更新计数的操作。或者，使用链接阻塞队列在线程之间传递“消息”（或者，在更高级的场景中，使用 Akka、Kafka、zeromq 等消息传递解决方案或其他类似解决方案）

Answer 3

由于您一次只添加一个元素，因此您也可以使用 Exchanger。看一下 Java 并发包。import java.io.IOException;import java.util.concurrent.Exchanger;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class Foo {&nbsp; &nbsp; public static void main(String[] args) throws IOException, InterruptedException {&nbsp; &nbsp; &nbsp; &nbsp; final Exchanger<Integer> exchanger = new Exchanger<>();&nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; ExecutorService executorService = Executors.newCachedThreadPool();&nbsp; &nbsp; &nbsp; &nbsp; executorService.execute(() -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // producer&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; int count = 0;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; while (!Thread.currentThread().isInterrupted()) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; exchanger.exchange(count++);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } catch (InterruptedException e) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Thread.currentThread().interrupt();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; });&nbsp; &nbsp; &nbsp; &nbsp; executorService.execute(() -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // reader&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; while (!Thread.currentThread().isInterrupted()) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; System.out.println("consume " + exchanger.exchange(null));&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } catch (InterruptedException e) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Thread.currentThread().interrupt();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; });&nbsp; &nbsp; &nbsp; &nbsp; System.in.read();&nbsp; &nbsp; &nbsp; &nbsp; executorService.shutdownNow();&nbsp; &nbsp; &nbsp; &nbsp; executorService.awaitTermination(10, TimeUnit.SECONDS);&nbsp; &nbsp; &nbsp; &nbsp; System.out.println("Shut down");&nbsp; &nbsp; }}