Java多进程生产者-消费者模型-socket通信

跨进程生产者-消费者模型

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