Остановка потоков обработки данных при помощи "Отравления очереди"

package com.company;
 
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
 
class Data {
    private int[] data;
 
    public Data(int[] data) {
        this.data = data;
    }
 
    int[] get() {
        return data;
    }
}
 
class PoisonArrayBlockingQueue<T> {
    private ArrayBlockingQueue<T> queue;
    private T poison;
 
    public PoisonArrayBlockingQueue(T poison) {
        if(poison == null) {
            throw new NullPointerException();
        }
        this.poison = poison;
        this.queue = new ArrayBlockingQueue<>(8);
    }
 
    T get() throws InterruptedException {
        return queue.take();
    }
 
    void put(T item) throws InterruptedException {
        queue.put(item);
    }
 
    boolean isPoison(T testObj) {
        return testObj == poison;
    }
 
    void addPoison() throws InterruptedException {
        put(poison);
    }
}
 
class ReadTask extends Thread {
    private PoisonArrayBlockingQueue<Data> store;
    private int id;
 
    public ReadTask(int id, PoisonArrayBlockingQueue<Data> store) {
        this.store = store;
        this.id = id;
    }
 
    @Override
    public void run() {
        System.out.println("ReadTask with id="+id+" is run");
        try {
            Data data = store.get();
            while( !store.isPoison(data) ) {
                System.out.println("Reader #"+id+" "+Arrays.toString(data.get()));
                Thread.sleep(1000);
                data = store.get();
            }
            store.addPoison();      // ADD POISON BACK to other thread
        } catch (InterruptedException e) {
            System.out.println("Thread Reader interrupt");
        }
        System.out.println("Read task #"+id+" finish");
    }
}
 
class WriteTask extends Thread {
    private PoisonArrayBlockingQueue<Data> store;
    private int id;
    private int countData;
 
    public WriteTask(int id, PoisonArrayBlockingQueue<Data> store, int countData) {
        this.store = store;
        this.id = id;
        this.countData = countData;
    }
 
    @Override
    public void run() {
        try {
            System.out.println("WriteTask with id="+id+" is run");
            for( int i=0; i<countData; ++i ) {
                int count = 3;      //1+(int)(Math.random()*5);
                int[] data = new int[count];
                for (int j=0; j<count; ++j) {
                    data[j] = (int)(Math.random()*10);
                }
                System.out.println("Writer #"+id+" "+Arrays.toString(data));
                store.put(new Data(data));
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            System.out.println("Thread Writer interrupt");
        }
        System.out.println("Writer task #"+id+" finish");
    }
}
 
public class Main {
    public static void main(String[] args) {
        final int READ_THREAD_COUNT = 2;
        final int WRITE_THREAD_COUNT = 3;
 
        Data poison = new Data(null);
        PoisonArrayBlockingQueue<Data> store = new PoisonArrayBlockingQueue<>(poison);
 
        List<Thread> readWaitList = new ArrayList<>(READ_THREAD_COUNT);
        List<Thread> writeWaitList = new ArrayList<>(WRITE_THREAD_COUNT);
        try {
            for (int i = 0; i < WRITE_THREAD_COUNT; ++i) {
                Thread t = new WriteTask(i, store, (int) (Math.random() * 4 + 2));
                writeWaitList.add(t);
                t.start();
            }
            for (int i = 0; i < READ_THREAD_COUNT; ++i) {
                Thread t = new ReadTask(i, store);
                readWaitList.add(t);
                t.start();
            }
 
            // WAIT OTHER THREADS
            for (int i = 0; i < WRITE_THREAD_COUNT; ++i) {
                writeWaitList.get(i).join();
                store.addPoison();   // <----ADD POISON-----
            }
            for (int i = 0; i < READ_THREAD_COUNT; ++i) {
                readWaitList.get(i).join();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("main() final execution");
    }
}