在实际的软件开发过程中,经常会碰到如下场景:某个模块负责产生数据,这些数据由另一个模块来负责处理(此处的模块是广义的,可以是类、函数、线程、进程等)。产生数据的模块,就形象地称为生产者;而处理数据的模块,就称为消费者。
单单抽象出生产者和消费者,还够不上是生产者/消费者模式。该模式还需要有一个缓冲区处于生产者和消费者之间,作为一个中介。生产者把数据放入缓冲区,而消费者从缓冲区取出数据。
1、生产者
随机生成一个大于20亿的正整数
2、消费者
判断某个数字是否素数(判断素数算法自己百度)
3、缓冲区
使用队列(FIFO)
public class Consumer implements Runnable {BlockingQueue<Integer> blockingQueue;int n;CountDownLatch countDownLatch;public Consumer(BlockingQueue<Integer> blockingQueue, int n, CountDownLatch countDownLatch) {this.blockingQueue = blockingQueue;this.n = n;this.countDownLatch=countDownLatch;}@Overridepublic void run() {for(int i=0;i<n;i++){try {int cur=blockingQueue.take();isPrime(cur);/* System.out.println("消费"+blockingQueue.size());*/} catch (InterruptedException e) {e.printStackTrace();}}System.out.println("消费者完成");countDownLatch.countDown();}int isPrime(int n){ //返回1表示判断为质数,0为非质数,在此没有进行输入异常检测double n_sqrt;if(n==2 || n==3) return 1;if(n%6!=1 && n%6!=5) return 0;n_sqrt=Math.floor(Math.sqrt((float)n));for(int i=5;i<=n_sqrt;i+=6){if(n%(i)==0 | n%(i+2)==0) return 0;}return 1;}}
public class Model {public static void excute(int producerNum,int consumerNum,int num,CountDownLatch countDownLatch){BlockingQueue<Integer> blockingQueue=new LinkedBlockingQueue<>(num);for(int i=0;i<producerNum;i++){new Thread(new Producer(blockingQueue,num/producerNum,countDownLatch)).start();}for(int i=0;i<consumerNum;i++){new Thread(new Consumer(blockingQueue,num/consumerNum,countDownLatch)).start();}}public static void main(String[] args) {CountDownLatch countDownLatch=new CountDownLatch(6);long s=System.currentTimeMillis();excute(2,4,1000000,countDownLatch);try {countDownLatch.await();} catch (InterruptedException e) {e.printStackTrace();}System.out.println((double) (System.currentTimeMillis()-s)/1000);}
}
public class Producer implements Runnable{BlockingQueue<Integer> blockingQueue;int n;CountDownLatch countDownLatch;public Producer(BlockingQueue<Integer> blockingQueue, int n,CountDownLatch countDownLatch) {this.blockingQueue = blockingQueue;this.n = n;this.countDownLatch=countDownLatch;}@Overridepublic void run() {Random ra =new Random();for(int i=0;i<n;i++){try {blockingQueue.put(ra.nextInt(2000000000)+1);/* System.out.println("生产"+blockingQueue.size());*/} catch (InterruptedException e) {e.printStackTrace();}}System.out.println("生产者完成");countDownLatch.countDown();}
}