rpc-maven

MyRPC.iml

@@ -0,0 +1,2 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4" />

pom.xml

@@ -0,0 +1,53 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.wherewego</groupId>
+    <artifactId>my-rpc</artifactId>
+    <version>0.0.1-alpha</version>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>6</source>
+                    <target>6</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+    <dependencies>
+        <!--使用protobuf序列化-->
+        <dependency>
+            <groupId>com.dyuproject.protostuff</groupId>
+            <artifactId>protostuff-core</artifactId>
+            <version>1.0.7</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.dyuproject.protostuff</groupId>
+            <artifactId>protostuff-runtime</artifactId>
+            <version>1.0.7</version>
+        </dependency>
+        <!-- netty -->
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+            <version>4.1.45.Final</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+            <version>2.2.5.RELEASE</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>1.7.30</version>
+        </dependency>
+    </dependencies>
+
+</project>

src/main/java/com/wherewego/rpc/call/Call.java

@@ -0,0 +1,90 @@
+package com.wherewego.rpc.call;
+
+/**
+ * @Author:lbl
+ * @Date:Created in 12:15 2020/2/29
+ * @Modified By:
+ */
+public class Call {
+    private int id;
+    private boolean isBack;
+    private String serverName;
+    private String beanName;
+    private String interfaceName;
+    private String methodName;
+    private Class[] paramTypes;
+    private Object[] params;
+    private Object result;
+
+    public String getBeanName() {
+        return beanName;
+    }
+
+    public void setBeanName(String beanName) {
+        this.beanName = beanName;
+    }
+
+    public boolean isBack() {
+        return isBack;
+    }
+
+    public void setBack(boolean back) {
+        isBack = back;
+    }
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public String getServerName() {
+        return serverName;
+    }
+
+    public void setServerName(String serverName) {
+        this.serverName = serverName;
+    }
+
+    public String getInterfaceName() {
+        return interfaceName;
+    }
+
+    public void setInterfaceName(String interfaceName) {
+        this.interfaceName = interfaceName;
+    }
+
+    public String getMethodName() {
+        return methodName;
+    }
+
+    public void setMethodName(String methodName) {
+        this.methodName = methodName;
+    }
+
+    public Class[] getParamTypes() {
+        return paramTypes;
+    }
+
+    public void setParamTypes(Class[] paramTypes) {
+        this.paramTypes = paramTypes;
+    }
+
+    public Object[] getParams() {
+        return params;
+    }
+
+    public void setParams(Object[] params) {
+        this.params = params;
+    }
+
+    public Object getResult() {
+        return result;
+    }
+
+    public void setResult(Object result) {
+        this.result = result;
+    }
+}

src/main/java/com/wherewego/rpc/call/NULL.java

@@ -0,0 +1,11 @@
+package com.wherewego.rpc.call;
+
+/**
+ * @Author:lbl
+ * @Date:Created in 23:00 2020/2/29
+ * @Modified By:
+ */
+public interface NULL {
+    String nul = "null";
+
+}

src/main/java/com/wherewego/rpc/cilent/ClientConnect.java

@@ -0,0 +1,76 @@
+package com.wherewego.rpc.cilent;
+
+import com.wherewego.rpc.codec.CallCodec;
+import com.wherewego.rpc.config.RpcConfig;
+import com.wherewego.rpc.handler.CallbackHandler;
+import com.wherewego.rpc.handler.TcpClientHandler;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import io.netty.handler.codec.LengthFieldPrepender;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ *用于客户端连接服务
+ * @Author:lbl
+ * @Date:Created in 17:18 2020/2/29
+ * @Modified By:
+ */
+@Component
+public class ClientConnect {
+    private static final Logger LOGGER = LoggerFactory.getLogger(ClientConnect.class);
+    //这个是用于数据读写的线程组，所有连接都由这个线程组来调度读写事件
+    EventLoopGroup group = new NioEventLoopGroup();
+    @Autowired
+    private RpcConfig config;
+
+    /**
+     * 线程不用这个连接了，释放掉（有线程池的话就还给线程池）
+     * @param channel
+     */
+    public void release(Channel channel){
+        channel.close();
+    }
+    /**
+     * 获得一个连接，因为有可能不同服务由不同的提供者提供，那连接是不一样的，所有要用服务名称来区分
+     * 怎么区分就扯到服务注册发现机制了，由注册中心告诉你哪些服务在哪些地址上
+     * 知道了服务名称，就能获取到服务提供者的连接（可能有多个，用负载均衡算法取其中一个）
+     * @param serverName 需要获取的服务名称
+     * @return
+     */
+    public Channel channel(String serverName, final CallbackHandler callback){
+        Bootstrap client = new Bootstrap();
+
+        client.group(group);
+        client.channel(NioSocketChannel.class);
+        //给NIoSocketChannel初始化handler， 处理读写事件
+        client.handler(new ChannelInitializer<NioSocketChannel>() {  //通道是NioSocketChannel
+            @Override
+            protected void initChannel(NioSocketChannel ch) throws Exception {
+                ChannelPipeline pipeline = ch.pipeline();
+                pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
+                pipeline.addLast(new LengthFieldPrepender(4));
+                // 序列化工具
+                pipeline.addLast(new CallCodec());
+                ch.pipeline().addLast(new TcpClientHandler(callback));
+            }
+        });
+
+        LOGGER.info("客户端建立连接");
+        //连接服务器
+        try {
+            ChannelFuture future = client.connect(config.getRemoteHost(), config.getRemotePort()).sync();
+
+            return future.channel();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e.getMessage());
+        }
+
+    }
+
+}

src/main/java/com/wherewego/rpc/cilent/processor/MyRPCBeanPostProcessor.java

@@ -0,0 +1,57 @@
+package com.wherewego.rpc.cilent.processor;
+
+import com.wherewego.rpc.cilent.proxy.ProxyFactory;
+import com.wherewego.rpc.config.annotation.Reference;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.stereotype.Component;
+
+import java.lang.reflect.Field;
+
+/**
+ * bean初始化前后执行处理
+ * @Author:lbl
+ * @Date:Created in 15:06 2020/2/29
+ * @Modified By:
+ */
+@Component
+public class MyRPCBeanPostProcessor implements BeanPostProcessor {
+    @Autowired
+    private ProxyFactory proxyFactory;
+    @Override
+    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+        // TODO Auto-generated method stub
+        Class<?> clz;
+        if(AopUtils.isAopProxy(bean)){
+            clz = AopUtils.getTargetClass(bean);
+        }else{
+            clz = bean.getClass();
+        }
+        try {
+            for (Field field:clz.getDeclaredFields()){
+                //找到有com.wherewego.rpc.annotation.Reference注解的变量
+                Reference reference = field.getAnnotation(Reference.class);
+                if(reference!=null){
+                    System.out.println("postProcessBeforeInitialization..."+reference.value());
+                    boolean access = field.isAccessible();
+
+                    field.setAccessible(true);
+                    //使用动态代理创建bean
+                    field.set(bean, proxyFactory.getProxy(field.getType(),reference.value()));
+                    field.setAccessible(access);
+                }
+            }
+        } catch (IllegalAccessException e) {
+            throw new RuntimeException(e.getMessage());
+        }
+        return bean;
+    }
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        // TODO Auto-generated method stub
+        return bean;
+    }
+}