Java多进程生产者-消费者模型-socket通信
程序员文章站
2022-03-10 08:38:42
跨进程生产者-消费者模型1、生产者随机生成一个大于20亿的正整数2、消费者判断某个数字是否素数3、缓冲区使用阻塞队列(LinkedBlockingQueue)要求1.使用n个生产者按要求生成1000000(100万)个数字,同时由m个消费者判断所生成的数字是否素数。记下所需时间。2.生产者、缓存项目和消费者分别各自独立项目,以三个进程独立启动(可运行在同一台机器内,也运行可以在不同机器中通过网络连接)源码注:同台机器上IDEA运行测试无误服务端package tcp;impor...
跨进程生产者-消费者模型
1、生产者
随机生成一个大于20亿的正整数
2、消费者
判断某个数字是否素数
3、缓冲区
使用阻塞队列(LinkedBlockingQueue)
要求
1.使用n个生产者按要求生成1000000(100万)个数字,同时由m个消费者判断所生成的数字是否素数。记下所需时间。
2.生产者、缓存项目和消费者分别各自独立项目,以三个进程独立启动(可运行在同一台机器内,也运行可以在不同机器中通过网络连接)
源码
注:同台机器上IDEA运行测试无误
服务端
package tcp;
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class Queue {
public static LinkedBlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();
public static void main(String[] args) throws IOException {
ServerSocket ss = new ServerSocket(60000);
ExecutorService pool = Executors.newFixedThreadPool(100);
while(true){
Socket socket= ss.accept();
Runnable runnable = new Runnable() {
@Override
public void run() {
try {
InputStream is = socket.getInputStream();
BufferedReader in = new BufferedReader(new InputStreamReader(is));
PrintWriter out=new PrintWriter(socket.getOutputStream());
while(true){
String s1 = in.readLine();
if(s1==null)break;
String[] list = s1.split(",");
if(list[0].equals("p")) {
blockingQueue.put(Integer.parseInt(list[1]));
}else if(list[0].equals("c")) {
out.println(blockingQueue.take());
out.flush();
}
}
System.out.println("队列长度:"+blockingQueue.size());
is.close();
socket.close();
}catch (Exception e) {
e.printStackTrace();
}
}
};
pool.submit(runnable);
}
}
}
生产者(客户端)
package tcp;
import java.io.IOException;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Producer {
public static Random rand = new Random();
public static int randNum=0;
public static int count = 0;
public static int num=4;//The Count Of Producer
public static void main(String[] args) throws UnknownHostException, IOException {
ExecutorService pool = Executors.newFixedThreadPool(num);
long start = System.currentTimeMillis();
while(count<50){
count++;
Runnable runnable = new Runnable() {
Socket socket= new Socket("127.0.0.1",60000);
PrintWriter out=new PrintWriter(socket.getOutputStream());
@Override
public void run() {
try {
for(int i=0; i<20000; i++){
randNum = 2000000000+(Math.abs(rand.nextInt(10000)));
String s = "p,"+randNum;
out.println(s);
out.flush();
}
socket.close();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
};
pool.submit(runnable);
}
pool.shutdown();
while(!pool.isTerminated());
System.out.println("生产者完工!");
System.out.println("实验耗时:"+(((System.currentTimeMillis())-start)/1000)+"s");
}
}
消费者(客户端)
package tcp;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Consumer {
public static int count = 0;
public static int num=4;The Count Of Consumer
public static ExecutorService pool = Executors.newFixedThreadPool(num);
public static void main(String[] args)throws UnknownHostException, IOException {
long start = System.currentTimeMillis();
while(count<50) {
count++;
Runnable runnable = new Runnable() {
Socket socket= new Socket("127.0.0.1",60000);
PrintWriter out=new PrintWriter(socket.getOutputStream());
BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
@Override
public void run() {
try {
for(int i=0; i<20000; i++){
String s = "c";
out.println(s);
out.flush();
//读数
String s1 = in.readLine();
//System.out.println(s1);
if(s1==null)break;
//判断素数
int num = Integer.parseInt(s1);
for (int i1 = 2; i1 <= Math.sqrt(num); i1++) {
if (num % i1 == 0 ) {
break;
}
}
}
socket.close();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
};
pool.submit(runnable);
}
pool.shutdown();
while(!pool.isTerminated());
System.out.println("消费者完工!");
System.out.println("实验耗时:"+(((System.currentTimeMillis())-start)/1000)+"s");
}
}
运行结果截图
生产者耗时
消费者耗时
本文地址:https://blog.csdn.net/weixin_44896629/article/details/112254186
上一篇: 【HuoLe的学习笔记】--数组模拟队列
下一篇: Java基础详解之包装类的装箱拆箱