Dubbo
Dubbo最早的定位是rpc框架，即远程服务调用，解决的是跨服务之间的方法调用问题，本文还是在这个定位基础上尝试手写一个简单的Dubbo
需求
首先要搭建测试的项目结构，两个服务consumer和provider，分别代表调用方和提供方，二者功能依赖于interface，其中暴露接口
interface包中定义一个接口
// interface
public interface HelloService {
String sayHello(String name);
}
provider实现
// provider
public class HelloServiceImpl implements HelloService {
public String sayHello(String name) {
return "hello "+name;
}
}
consumer调用
// consumer
public class Consumer {
public static void main(String[] args) {
// todo 获取不到HelloService的实现
HelloService helloService = null;
System.out.println(helloService.sayHello("pq"));
}
}
当前的需求即consumer服务调用provider服务里sayHello方法的实现，显然当前无法实现，这是一种远程发放调用，我们在新建一个Module命名为dubbo，意图通过依赖它来实现远程方法的调用
dubbo
网络
由于跨服务了，所以远程调用必然是要走网络的，dubbo使用了netty，我们也用netty来实现通讯
首先定义网络请求的数据，远程调用需要的信息：哪个类，哪个方法，什么参数，我们把这些信息封装一下
// dubbo
@Data
@AllArgsConstructor
public class Invocation implements Serializable {
private String className;
private String methodName;
private Class<?>[] paramTypes;
private Object[] args;
}
服务端
provider作为服务的提供方，需要依靠netty搭建一个服务器，当接受到请求(Invocation对象)时，可以根据className，methodName等信息找到对应的本地方法进行调用
所以provider首先要维护一个map存储className和class的对应关系，这样在收到请求时可以通过className找到对应的类，再通过反射获取对应的方法进行调用
在我们的dubbo框架中封装这么一个map结构供provider使用
// dubbo
public class LocalRegister {
private static Map<String, Object> map = new HashMap<String, Object>();
public static void register(String className, Object impl) {
map.put(className, impl);
}
public static Object get(String className) {
return map.get(className);
}
}
然后再做一个处理请求netty服务供provider使用
// dubbo
public class NettyServer {
public void start(Integer port) {
try {
final ServerBootstrap bootstrap = new ServerBootstrap();
EventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("bossGroup", true));
EventLoopGroup workerGroup = new NioEventLoopGroup(10, new DefaultThreadFactory("workerGroup", true));
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast("decoder", new ObjectDecoder(ClassResolvers
.weakCachingConcurrentResolver(this.getClass()
.getClassLoader())));
channel.pipeline().addLast("encoder", new ObjectEncoder());
channel.pipeline().addLast("handler", new RequestHandler());
}
});
ChannelFuture cf = bootstrap.bind(port).sync();
cf.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
对应的handler如下
// dubbo
public class RequestHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Invocation invocation = (Invocation) msg;
// 根据className获取寄存的服务对象
Object serviceImpl = LocalRegister.get(invocation.getClassName());
// 通过methodName等信息获取对应的方法
Method method = serviceImpl.getClass().getMethod(invocation.getMethodName(), invocation.getParamTypes());
// 调用方法
Object result = method.invoke(serviceImpl, invocation.getArgs());
// 返回服务结果
ctx.writeAndFlush(result);
}
}
provider启动类Starter
// provider
public class Starter {
public static void main(String[] args) {
// 存储服务于名字映射关系
HelloServiceImpl helloService = new HelloServiceImpl();
String className = HelloService.class.getName();
LocalRegister.register(className, helloService);
// 开启netty服务
NettyServer nettyServer = new NettyServer();
System.out.println("provider 端口号9001");
nettyServer.start(9001);
}
}
代理
consumer只能拿到到HelloService接口，那么实例化的方法可以采用jdk动态代理生成代理实现，而代理的实际执行方式是通过netty网络发送请求给provider，
首先还是在dubbo框架中封装一个netty的客户端供consumer发起请求
// dubbo
@Setter
public class NettyClient {
/**
* 管道上下文
*/
private volatile ChannelHandlerContext channelHandlerContext;
/**
* 返回消息暂存
*/
private Object message;
public void start(String hostName, Integer port) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
channel.pipeline().addLast("decoder", new ObjectDecoder(ClassResolvers
.weakCachingConcurrentResolver(this.getClass()
.getClassLoader())));
channel.pipeline().addLast("encoder", new ObjectEncoder());
channel.pipeline().addLast(new ResponseHandler(NettyClient.this));
}
});
bootstrap.connect(hostName, port).sync();
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 发送远程调用
* @param hostName
* @param port
* @param invocation
* @return
*/
public synchronized String send(String hostName, Integer port, Invocation invocation) {
start(hostName, port);
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 发送数据
channelHandlerContext.writeAndFlush(invocation);
// 等待
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 返回数据
return message.toString();
}
}
其中的ResponseHandler入下
// dubbo
public class ResponseHandler extends ChannelInboundHandlerAdapter {
private final NettyClient client;
public ResponseHandler(NettyClient client) {
this.client = client;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
synchronized (client) {
client.notify();
}
client.setChannelHandlerContext(ctx);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
client.setMessage(msg);
synchronized (client) {
client.notify();
}
}
}
然后在我们的dubbo框架中实现创建代理
// dubbo
public class ProxyFactory {
/**
* 根据接口创建代理 jdk动态代理
* @param interfaceClass
* @param <T>
* @return
*/
public static <T> T getProxy(final Class<T> interfaceClass) {
return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{interfaceClass}, new InvocationHandler() {
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// 请求封装成对象
Invocation invocation = new Invocation(interfaceClass.getName(), method.getName(), method.getParameterTypes(), args);
NettyClient nettyClient = new NettyClient();
// 发起网络请求
String response = nettyClient.send("127.0.0.1", 9001, invocation);
return response;
}
});
}
}
最后回到consumer添加启动类，通过代理创建HelloService的实现，尝试调用provider的sayHello方法
// consumer
public class Consumer {
public static void main(String[] args) {
HelloService helloService = ProxyFactory.getProxy(HelloService.class);
System.out.println(helloService.sayHello("pq"));
}
}
测试
- 启动
provider，输出如下
provider
- 启动
consumer，输出如下
consumer
证明已实现跨远程方法调用~
网友评论