欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

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");
    }
}

运行结果截图

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

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

本文地址:https://blog.csdn.net/weixin_44896629/article/details/112254186