public interface IHelloService {
String sayHello(String content);
String saveUser(User user);
}public class RpcRequest implements Serializable {
private static final long serialVersionUID = 1041322970832269647L;
private String className;
private String methodName;
private Object[] parameters;
public String getClassName() {
return className;
}
public void setClassName(String className) {
this.className = className;
}
public String getMethodName() {
return methodName;
}
public void setMethodName(String methodName) {
this.methodName = methodName;
}
public Object[] getParameters() {
return parameters;
}
public void setParameters(Object[] parameters) {
this.parameters = parameters;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
RpcRequest that = (RpcRequest) o;
return Objects.equals(className, that.className) && Objects.equals(methodName, that.methodName) && Arrays.equals(parameters, that.parameters);
}
@Override
public int hashCode() {
int result = Objects.hash(className, methodName);
result = 31 * result + Arrays.hashCode(parameters);
return result;
}
}public class User {
private String name;
private int age;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
@Override
public String toString() {
return "User{" +
"name='" + name + '\'' +
", age=" + age +
'}';
}
public void setAge(int age) {
this.age = age;
}
}public class HelloServiceImpl implements IHelloService{
@Override
public String sayHello(String content) {
System.out.println("Request in: " + content);
return "Say Hello";
}
@Override
public String saveUser(User user) {
System.out.println("Request in: " + user);
return "SUCCESS";
}
}public class ProcessorHandler implements Runnable{
private Socket socket;
private Object service;
public ProcessorHandler(Socket socket, Object service) {
this.socket = socket;
this.service = service;
}
@Override
public void run() {
ObjectInputStream objectInputStream = null;
ObjectOutputStream objectOutputStream = null;
try {
objectInputStream = new ObjectInputStream(socket.getInputStream());
// 请求那个类
RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject();
Object result = invoke(rpcRequest);
objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
objectOutputStream.writeObject(result);
objectOutputStream.flush();
} catch (IOException | ClassNotFoundException e) {
e.printStackTrace();
} catch (InvocationTargetException e) {
e.printStackTrace();
} catch (NoSuchMethodException e) {
e.printStackTrace();
} catch (IllegalAccessException e) {
e.printStackTrace();
} finally {
if (objectInputStream != null) {
try {
objectInputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if (objectOutputStream != null) {
try {
objectOutputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
private Object invoke(RpcRequest rpcRequest) throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, IllegalAccessException {
// 反射调用
Object[] args = rpcRequest.getParameters(); // 拿到客户端的参数
Class<?>[] types = new Class[args.length]; // 获得每个参数的类型
for (int i = 0; i < types.length; i++) {
types[i] = args[i].getClass();
}
Class clazz = Class.forName(rpcRequest.getClassName());
//clazz.getMethods(); // sayHello saveUser
Method method = clazz.getMethod(rpcRequest.getMethodName(), types);
Object result = method.invoke(service, args);
return result;
}
}public class RpcProxyServer {
ExecutorService executorService = Executors.newCachedThreadPool();
public void publisher(Object service, int port) {
ServerSocket serverSocket = null;
try {
serverSocket = new ServerSocket(port);
while (true) {
Socket socket = serverSocket.accept();
executorService.execute(new ProcessorHandler(socket, service));
}
} catch (IOException e) {
e.printStackTrace();
} finally {
if (serverSocket != null) {
try {
serverSocket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}public class App {
public static void main(String[] args) {
IHelloService helloService = new HelloServiceImpl();
RpcProxyServer proxyServer = new RpcProxyServer();
proxyServer.publisher(helloService, 8080);
}
}public class App {
public static void main(String[] args) {
System.out.println("Hello World");
RpcProxyClient rpcProxyClient = new RpcProxyClient();
IHelloService iHelloService = rpcProxyClient.clientProxy
(IHelloService.class, "localhost", 8080);
String result = iHelloService.sayHello("MoreMind");
System.out.println(result);
}
}public class RpcProxyClient {
public <T> T clientProxy(final Class<T> interfaceCls, final String host, final int port) {
return (T) Proxy.newProxyInstance(interfaceCls.getClassLoader(),
new Class<?>[]{interfaceCls}, new RemoteInvocationHandler(host, port));
}
}public class RemoteInvocationHandler implements InvocationHandler {
private String host;
private int port;
public RemoteInvocationHandler(String host, int port) {
this.host = host;
this.port = port;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
System.out.println("com in");
RpcRequest rpcRequest = new RpcRequest();
rpcRequest.setClassName(method.getDeclaringClass().getName());
rpcRequest.setMethodName(method.getName());
rpcRequest.setParameters(args);
RpcNetTransport netTransport = new RpcNetTransport(host, port);
Object result = netTransport.send(rpcRequest);
return result;
}
}public class RpcNetTransport {
private String host;
private int port;
public RpcNetTransport(String host, int port) {
this.host = host;
this.port = port;
}
public Object send(RpcRequest rpcRequest) {
Socket socket = null;
Object result = null;
ObjectOutputStream objectOutputStream = null;
ObjectInputStream objectInputStream = null;
try {
socket = new Socket(host, port);// 建立链接
objectOutputStream = new ObjectOutputStream(socket.getOutputStream()); // 网络通信
objectOutputStream.writeObject(rpcRequest); // 序列化
objectOutputStream.flush();
objectInputStream = new ObjectInputStream(socket.getInputStream());
result = objectInputStream.readObject();
return result;
} catch (IOException | ClassNotFoundException e) {
e.printStackTrace();
} finally {
if (objectInputStream != null) {
try {
objectInputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if (objectOutputStream != null) {
try {
objectOutputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return result;
}
}@Target(ElementType.TYPE) //类或接口
@Retention(RetentionPolicy.RUNTIME)
@Component //被Spring扫描
public @interface RpcService {
Class<?> value(); //拿到服务的接口
String version() default "";
}@Component(value = "mRpcServer")
public class MRpcServer implements ApplicationContextAware, InitializingBean {
ExecutorService executorService = Executors.newCachedThreadPool();
private Map<String,Object> handlerMap=new HashMap();
private int port;
public MRpcServer(int port) {
this.port = port;
}
@Override
public void afterPropertiesSet() throws Exception {
ServerSocket serverSocket=null;
try {
serverSocket=new ServerSocket(port);
while(true){//不断接受请求
Socket socket=serverSocket.accept();//BIO
//每一个socket 交给一个processorHandler来处理
executorService.execute(new ProcessorHandler(socket,handlerMap));
}
} catch (IOException e) {
e.printStackTrace();
}finally {
if(serverSocket!=null){
try {
serverSocket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(RpcService.class);
if (!serviceBeanMap.isEmpty()) {
for (Object serviceBean : serviceBeanMap.values()) {
RpcService rpcService = serviceBean.getClass().getAnnotation(RpcService.class);
String serviceName = rpcService.value().getName();
String version = rpcService.version();
if (!StringUtils.isEmpty(version)) {
serviceName += "-" + version;
}
handlerMap.put(serviceName, serviceBean);
}
}
}
}@RpcService(value = IHelloService.class, version = "v1.0")
public class HelloServiceImpl implements IHelloService{
@Override
public String sayHello(String content) {
System.out.println("【V1.0】request in sayHello:"+content);
return "【V1.0】Say Hello:"+content;
}
@Override
public String saveUser(User user) {
System.out.println("【V1.0】request in saveUser:"+user);
return "【V1.0】SUCCESS";
}
}@Configuration
@ComponentScan(basePackages = "cn.moremind")
public class SpringConfig {
@Bean(name="mRpcServer")
public MRpcServer mRpcServer(){
return new MRpcServer(8080);
}
}public class App {
public static void main(String[] args) {
ApplicationContext context = new AnnotationConfigApplicationContext(SpringConfig.class);
((AnnotationConfigApplicationContext) context).start();
RpcProxyClient rpcProxyClient = context.getBean(RpcProxyClient.class);
IHelloService iHelloService = rpcProxyClient.clientProxy
(IHelloService.class, "localhost", 8080);
String result = iHelloService.sayHello("MoreMind");
System.out.println(result);
}
}public class RemoteInvocationHandler implements InvocationHandler {
private String host;
private int port;
public RemoteInvocationHandler(String host, int port) {
this.host = host;
this.port = port;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
System.out.println("com in");
RpcRequest rpcRequest = new RpcRequest();
rpcRequest.setClassName(method.getDeclaringClass().getName());
rpcRequest.setMethodName(method.getName());
rpcRequest.setParameters(args);
rpcRequest.setVersion("v2.0");
RpcNetTransport netTransport = new RpcNetTransport(host, port);
Object result = netTransport.send(rpcRequest);
return result;
}
}@Configuration
public class SpringConfig {
@Bean(name = "rpcProxyClient")
public RpcProxyClient rpcProxyClient() {
return new RpcProxyClient();
}
}
