分布式Web应用----基于Socket+动态代理实现简单RPC 生产者消费者模型

àì夳堔傛蜴生んèń 2022-08-21 04:41 22阅读 0赞

原文连接：[https://ksfmaster.com/article/9][https_ksfmaster.com_article_9]  
\#\#写在前面  
前面一文主要简单介绍了JAVA动态代理基本原理，这也是实现RPC的基本知识，这里我们运用Socket简单实现一个远程过程调用，方便大家理解RPC的基本原理，希望对大家有所帮助。

\#\#新建People接口类与Man实现类  
**接口类**

public interface People {
        public  String sayHello(String name);
    }

**实现类**

public class Man implements People {
        public String sayHello(String name) {
            System.out.println("call Man sayHello ");
            return "Hello "+name;
        }
    }

这两个类都比较简单，没什么好说的了。

\#\#新建生产者线程类ProducerTask 和处理Socket服务类RPCProducer

**生产者处理连接线程类**

public class ProducerTask implements  Runnable {
    
        Socket client=null;
        public  ProducerTask(Socket client){
            this.client=client;
        }
        public void run() {
            ObjectInputStream input=null;
            ObjectOutputStream output=null;
            try {
                input=new ObjectInputStream(client.getInputStream());
                String interFaceName=input.readUTF();
                Class<?> service=Class.forName(interFaceName);
                String methodName=input.readUTF();
                Class<?> [] paramterTypes= (Class<?>[]) input.readObject();
                Object [] arguments= (Object[]) input.readObject();
                Method method=service.getMethod(methodName,paramterTypes);
                Object result=method.invoke(service.newInstance(),arguments);
                System.out.println("远程调用对象名为："+service.getName());
                System.out.println("远程调用方法名为："+methodName);
                for (Object obj:arguments){
                    System.out.println("远程调用参数为:"+obj);
                }
                output=new ObjectOutputStream(client.getOutputStream());
                output.writeObject(result);
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                try {
                  output.close();
                   input.close();
                   client.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

ProducerTask 类主要是处理请求子线程，在该类中通过Socket连接获取消费者发送过来的 对象名、方法名、方法参数、参数列表，再利用反射机制，完成方法的调用，并向消费者返回调用的结果。

**处理Socket服务类**

public class RPCProducer {
        static Executor executor= Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        public static void exproter(String hostName,int port) throws Exception {
            ServerSocket server=new ServerSocket();
            server.bind(new InetSocketAddress(hostName,port));
            try {
                while (true){
                    System.out.println("get client call");
                    executor.execute(new ProducerTask(server.accept()));
                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                server.close();
            }
        }
    }

该类是生产者监听连接类，这里面主要使用了线程池，每当一个连接过来时，将线程池中增加一个线程，并让子线程去处理连接请求。

\#\#新建消费者动态代理类和消费者类  
**消费者动态代理类**

public class ConsumerHandler<T> implements InvocationHandler {
        private Class<?> object;
        private InetSocketAddress address;
        public ConsumerHandler(Class<?> object, InetSocketAddress address){
            this.object=object;
            this.address=address;
        }
    
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            System.out.println(" call peopleProxyHandler invoke ");
            System.out.println("调用对象为："+object.getClass().getName());
            System.out.println("调用方法为："+method.getName());
            for (Object obj:args){
                System.out.println("参数为:"+obj.toString());
            }
    
            Class cl=Class.forName(object.getClass().getName());
    
            Socket socket=null;
            ObjectOutputStream output=null;
            ObjectInputStream  input=null;
    
            socket=new Socket();
            socket.connect(address);
            output=new ObjectOutputStream(socket.getOutputStream());
            output.writeUTF(object.getName());
            output.writeUTF(method.getName());
            output.writeObject(method.getParameterTypes());
            output.writeObject(args);
            input =new ObjectInputStream(socket.getInputStream());
    
            return  input.readObject();
        }
    }

该类实现了InvocationHandler 接口，主要将对本地方法的调用，代理到此类中，在invoke方法中，获取调用方法的对象名、方法名、参数类型及参数列表，再通过连接生产者的Socket服务，将这些参数传送给生产者，并返回生产者返回的结果。

**消费者类**

public class RPCConsumer<T> {
        public T improter(Class<?> serviceClass, InetSocketAddress address){
            ConsumerHandler<T> consumerHandler=new ConsumerHandler<T>(serviceClass,address);
            Class<?> intf=serviceClass.getInterfaces()[0];
            Class<?> [] interfaces=new Class<?>[]{intf};
            return (T) Proxy.newProxyInstance(serviceClass.getClassLoader(),interfaces,consumerHandler);
        }
    }
    
    该类主要是返回一个代理对象，完成动态代理功能

\#\#新建生产者和服务者测试类

**生产者测试类**

public class ProducerTest {
        public  static  void  main(String [] rags){
            new Thread(new Runnable() {
                public void run() {
                    try {
                        RPCProducer.exproter("localhost",8080);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

这是生产者测试类，主要在开启生产者服务，监听端口为8080

**消费者测试类**

public static void main(String [] rags){
            RPCConsumer<People> peopleRPCConsumer=new RPCConsumer<People>();
            People people=peopleRPCConsumer.improter(Man.class,new InetSocketAddress("localhost",8080));
            System.out.println(people.sayHello("zhaochao"));
    
        }

消费者测试类，通过peopleRPCConsumer 生成代理对象people，调用people的方法，完成远程调用

\#\#结果

**生产者输出**

get client call
    get client call
    call Man sayHello 
    远程调用对象名为：com.zhaocaho.proxy.Man
    远程调用方法名为：sayHello
    远程调用参数为:zhaochao

**消费者输出**

call peopleProxyHandler invoke 
    调用对象为：java.lang.Class
    调用方法为：sayHello
    参数为:zhaochao
    Hello zhaochao

[https_ksfmaster.com_article_9]: https://ksfmaster.com/article/9