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

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以唤醒另一个线程，并且本身应进入等待状态。

 public synchronized void writeList() {

      while (true) {

            if (flag) {

                count++;

                list.add(count);

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

                System.out.println(list);

                flag = false;

                try {

                    notify(); //notify the read thread that write is complete

                    wait();   // go into the waiting state so that no further write is done until the current element is removed by the read thread.

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

            } else {

                try {

                    wait(); //wait in else, otherwise while will run endlessly

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

           }

      }

 }

 public synchronized void readList() {

        while (true) {

            System.out.println("in read");

            if (!flag) {

                Integer i = list.remove(0);

                System.out.println(i + "..removed at index by" + Thread.currentThread().getName());

                flag = true;

                try {

                    notify(); //notify write thread that read is complete

                    wait();   //go into wait until new element is inserted

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

            } else {

                try {

                    wait(); //wait in else otherwise while runs endlessly

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

            }

        }

    }

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 {

    public static void main(String[] args) throws IOException, InterruptedException {

        final Exchanger<Integer> exchanger = new Exchanger<>();    

        ExecutorService executorService = Executors.newCachedThreadPool();

        executorService.execute(() -> {

            // producer

            int count = 0;

            while (!Thread.currentThread().isInterrupted()) {

                try {

                    exchanger.exchange(count++);

                } catch (InterruptedException e) {

                    Thread.currentThread().interrupt();

                }

            }

        });

        executorService.execute(() -> {

            // reader

            while (!Thread.currentThread().isInterrupted()) {

                try {

                    System.out.println("consume " + exchanger.exchange(null));

                } catch (InterruptedException e) {

                    Thread.currentThread().interrupt();

                }

            }

        });

        System.in.read();

        executorService.shutdownNow();

        executorService.awaitTermination(10, TimeUnit.SECONDS);

        System.out.println("Shut down");

    }

}