Semaphore和Exchanger实现生产者消费者模式

Semaphore实现生产者消费者模式

• acquire方法：获取信号量的许可，并把信号量的值减1
• release方法：释放一个许可，将信号量的值加1

package thread;

import java.util.LinkedList;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;

public class ProducerConsumer4 {
    static Semaphore emptySlot = new Semaphore(1);
    static Semaphore fullSlot = new Semaphore(0);

    public static void main(String[] args) throws InterruptedException {
        Queue<Integer> queue = new LinkedList<>();
        ProducerConsumer4.Producer p1 = new ProducerConsumer4.Producer(queue);
        ProducerConsumer4.Consumer c1 = new ProducerConsumer4.Consumer(queue);
        p1.start();
        c1.start();
        p1.join();
        c1.join();
    }

    static class Producer extends Thread {
        Queue<Integer> queue;

        Producer(Queue<Integer> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                emptySlot.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized (ProducerConsumer4.class) {
                int tmp = new Random().nextInt();
                queue.offer(tmp);
                System.out.println("Producing " + tmp);
            }
            fullSlot.release();
        }

    }

    static class Consumer extends Thread {
        Queue<Integer> queue;

        Consumer(Queue<Integer> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                fullSlot.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized (ProducerConsumer4.class) {
                System.out.println("Consuming " + queue.poll());
            }
            emptySlot.release();
        }
    }

}

Exchanger实现生产者消费者

• 线程到达交换点，进行交换

package thread;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Exchanger;

public class ProducerConsumer5 {

    public static void main(String[] args) throws InterruptedException {
        Exchanger<Integer> exchanger = new Exchanger<>();
        ProducerConsumer5.Producer p1 = new ProducerConsumer5.Producer(exchanger);
        ProducerConsumer5.Consumer c1 = new ProducerConsumer5.Consumer(exchanger);
        p1.start();
        c1.start();
        p1.join();
        c1.join();
    }

    static class Producer extends Thread {
        Exchanger<Integer> exchanger;

        Producer(Exchanger<Integer> exchanger) {
            this.exchanger = exchanger;
        }

        @Override
        public void run() {
            while (true) {
                int tmp = new Random().nextInt();
                try {
                    exchanger.exchange(tmp);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Producing " + tmp);
            }

        }

    }

    static class Consumer extends Thread {
        Exchanger<Integer> exchanger;

        Consumer(Exchanger<Integer> exchanger) {
            this.exchanger = exchanger;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    System.out.println("Consuming " + exchanger.exchange(null));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

        }
    }
}