一个简单的rpc框架的实现

chenkangxian

浏览: 87992 次
性别:
来自: 杭州

最近访客更多访客>>

夕沅的爱

御羽倾城

tdktdktdk

xiaomaqiqi8

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

分布式

rpc rpc框架 rpc中间件 SOA实现

为了降低开发成本，提升已有系统的利用率，企业往往会构建自己的SOA体系结构，SOA构建的手段有多种，可以通过webservice，restful，rmi，socket等等，笔者通过java socket，来构建一个简单的SOA体系结构，以下代码仅供参考。

1. 需要实现SOA的服务接口

package com.chenkangxian.rpc.impl;

/**
 *
 * @Author: chenkangxian
 *
 * @Annotation: 根据key取数据接口
 *
 * @Date:2012-5-13
 *
 * @Copyright: 2012 chenkangxian, All rights reserved.
 *
 */

public interface DataService {

    String getData(String key);

}

2. 接口的实现

package com.chenkangxian.rpc.impl;

/**
 *
 * @Author: chenkangxian
 *
 * @Annotation: 根据key取数据服务实现
 *
 * @Date:2012-5-13
 *
 * @Copyright: 2012 chenkangxian, All rights reserved.
 *
 */
public class DataServiceImpl implements DataService {

	public String getData(String key) {
		
        return "this is the data when key = " + key ;
    }

}

3. 执行代理

/**
 * 
 */
package com.chenkangxian.rpc.impl;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.Socket;

/**
 * @Author: chenkangxian
 *
 * @Annotation: 执行代理
 *
 * @Date:2012-5-15
 *
 * @Copyright: 2012 chenkangxian, All rights reserved.
 * 
 */
public class InvocationProxy implements InvocationHandler{

	private String host;
	private int port;

	public InvocationProxy(String host, int port){
		this.host = host;
		this.port = port;
	}

	public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable {
		Socket socket = new Socket(host, port);
		try {
			ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
			try {
				output.writeUTF(method.getName());
				output.writeObject(method.getParameterTypes());
				output.writeObject(arguments);
				ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
				try {
					Object result = input.readObject();
					if (result instanceof Throwable) {
						throw (Throwable) result;
					}
					return result;
				} finally {
					input.close();
				}
			} finally {
				output.close();
			}
		} finally {
			socket.close();
		}
	}
}

4. 服务消费者

package com.chenkangxian.rpc.impl;

/**
 * @Author: chenkangxian
 *
 * @Annotation: 服务消费者
 *
 * @Date:2012-5-13
 *
 * @Copyright: 2012 chenkangxian, All rights reserved.
 *
 */
public class RpcConsumer {
    
    public static void main(String[] args) throws Exception {
        DataService service = RpcFramework.refer(DataService.class, "127.0.0.1", 1234);
        for (int i = 0; i < Integer.MAX_VALUE; i ++) {
            String value = service.getData("key_" + i);
            System.out.println(value);
            Thread.sleep(1000);
        }
    }
    
}

5. 远程调用框架

package com.chenkangxian.rpc.impl;

import java.lang.reflect.Proxy;
import java.net.ServerSocket;
import java.net.Socket;

/**
 *
 * @Author: chenkangxian
 *
 * @Annotation: 简单的远程调用框架实现
 *
 * @Date:2012-5-13
 *
 * @Copyright: 2012 chenkangxian, All rights reserved.
 *
 */
public class RpcFramework {
	
	/**
	 * 暴露服务
	 * 
	 * Author: chenkangxian
	 *
	 * Last Modification Time: 2012-5-15
	 *
	 * @param service 服务实现
	 * @param port 服务端口
	 * @throws Exception
	 */
	public static void export(final Object service, int port) throws Exception {
		if (service == null)
			throw new IllegalArgumentException("service instance == null");
		if (port <= 0 || port > 65535)
			throw new IllegalArgumentException("Invalid port " + port);
		System.out.println("Export service " + service.getClass().getName() + " on port " + port);
		ServerSocket server = new ServerSocket(port);
		for(;;) {
			try {
				final Socket socket = server.accept();
				ThreadPoolHelp.getExecutorInstance().execute(new WorkThread(service, socket));
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 引用服务
	 * 
	 * Author: chenkangxian
	 *
	 * Last Modification Time: 2012-5-15
	 *
	 * @param <T> 接口泛型
	 * @param interfaceClass 接口类型
	 * @param host 服务器主机名
	 * @param port 服务器端口
	 * @return 远程服务
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	public static <T> T refer(final Class<T> interfaceClass, final String host, final int port) throws Exception {
		
		if (interfaceClass == null)
			throw new IllegalArgumentException("Interface class == null");
		if (! interfaceClass.isInterface())
			throw new IllegalArgumentException("The " + interfaceClass.getName() + " must be interface class!");
		if (host == null || host.length() == 0)
			throw new IllegalArgumentException("Host == null!");
		if (port <= 0 || port > 65535)
			throw new IllegalArgumentException("Invalid port " + port);
		
		System.out.println("Get remote service " + interfaceClass.getName() + " from server " + host + ":" + port);
		
		return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[] {interfaceClass}, new InvocationProxy(host,port));
	}

}

6. 服务提供者

package com.chenkangxian.rpc.impl;

/**
 * @Author: chenkangxian
 *
 * @Annotation: 服务提供者
 *
 * @Date:2012-5-13
 *
 * @Copyright: 2012 chenkangxian, All rights reserved.
 *
 */
public class RpcProvider {

    public static void main(String[] args) throws Exception {
        DataService service = new DataServiceImpl();
        RpcFramework.export(service, 1234);
    }

}

7. 线程池帮助类

/**
 * 
 */
package com.chenkangxian.rpc.impl;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Author: chenkangxian
 *
 * @Annotation:  线程池帮助类
 *
 * @Date:2012-5-15
 *
 * @Copyright: 2012 chenkangxian, All rights reserved.
 * 
 */
public class ThreadPoolHelp {

	private static ExecutorService executor ;
	
	static{
		executor = Executors.newFixedThreadPool(20);
	}
	
	public static ExecutorService getExecutorInstance(){
		return executor;
	}
}

8. 工作线程

/**
 * 
 */
package com.chenkangxian.rpc.impl;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.Socket;

/**
 *
 * @Author: chenkangxian
 *
 * @Annotation: 工作线程
 *
 * @Date:2012-5-15
 *
 * @Copyright: 2012 chenkangxian, All rights reserved.
 * 
 */
public class WorkThread implements Runnable {
	
	private Object service;
	
	private Socket socket;
	
	public WorkThread(Object service,Socket socket){
		this.service = service;
		this.socket = socket;
	}

    @Override
    public void run() {
        try {
            try {
                ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                try {
                    String methodName = input.readUTF();
                    Class<?>[] parameterTypes = (Class<?>[])input.readObject();
                    Object[] arguments = (Object[])input.readObject();
                    ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                    try {
                        Method method = service.getClass().getMethod(methodName, parameterTypes);
                        Object result = method.invoke(service, arguments);
                        output.writeObject(result);
                    } catch (Throwable t) {
                        output.writeObject(t);
                    } finally {
                        output.close();
                    }
                } finally {
                    input.close();
                }
            } finally {
                socket.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

分享到：

ThreadLocal的使用及实现 | zookeeper下的分布式锁

2012-05-15 21:58
浏览 4658
评论(2)
分类:企业架构
查看更多

2 楼 lizhi8888 2015-12-15

lizhi8888 写道

你好，我按你写的书中第一章的rpc实现，写了几个类，发现在provider在反序列话的时候，提示：Exception in thread "main" java.lang.ClassCastException: [Ljava.lang.Class; cannot be cast to java.lang.Class
at rpc.Provider.main(Provider.java:30)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140)

我觉得是是classloader不同导致的，但是用该博文提供的rpc实现确实可以运行的。这到底是为什么？

已经解决，是我写错了，Class [] 强制转换成 Class导致的

1 楼 lizhi8888 2015-12-15

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论