首页 微博热点正文

现在很多企业都在使用Dubbo或者Spring Cloud做企业的微服务架构,其实对于Dubbo最核心的技术就是RPC调用,现在我们就来动手自己编写一个RPC框架,通过这篇文章的学习,你将学习到

  • 分布式系统的概念
  • RPC远程方法调用的应用
  • Dubbo的原理深入理解

当然,如果要完全自己编写一个RPC框架,我们需要掌握以下知识点

  • 网络编程(网络通信) 本文将使用netty4网络通信框架
  • 多线程相关知识
  • 反射相关知识
  • jdk的动态代理
  • Spring框架的相关知识

如果对于上述的知识点有一部分不是很理解,也不会影响你阅读本文和对Dubbo的RPC调用原理的理解

好了,我们先来简单的描述一下整个RPC调用的业务流程图

rpc通信模型.png

为了可以实现上面的RPC调用,我们创建的RPC框架的模块之间的关系图如下:

RPC框架流程图.png

对于上面的每个模块的具体作用,使用一个表格简单的进行描述

模块名称主要功能rpc-register主要完成可注册中心Zookeeper的交互
RPC服务端使用该模块往注册中心注册地址和端口
RPC客户端通过该模块获取实时已近注册的服务地址和端口rpc-common定义RPC通信的请求消息和响应消息的规则,以及消息的序列化和反序列化的帮助类rpc-serverRPC服务端,启动RPC服务,扫描app-server中的所有可以提供的服务列表并保存杨会珍
接受RPC客户端的消息并且通过反射调用具体的方法

响应RPC客户端,把方法执行结果返回到RPC客户端rpc-clientRPC客户端,通过网络通信往RPC服务端发送请求调用消息

接受服务端的响应消息

配置动态代理类,所有的方法调用都通过网络调用发送到RPC服务端app-common具体的应用中的接口和JavaBean对象,类似于service模块和bean模块app-server通过Spring的配置启动SpringContext,并且配置RpcServer和RpcRegistry Bean对象的创建
实现app-common中的接口,并且在接口上添加注解@RpcService(IProductService.class)可以让RPCServer识别到该服务
启动服务app-client通过Spring的配置创建RpcDiscover对象和RpcProxy对象,其中RpcDiscover用于从注册中心获取到服务的地址信息,RpcProxy用于创建类的动态代理对象

接下来我们来看一下具体的实现代码

  1. rpc-register
  2. 这个模块用户和注册中心进行交互,主要包括三个类
  3. 具体的实现代码
package cn.wolfcode.rpc.register;public interface Constant { //定义客户端连接session会话超时时间,单位为毫秒,该值的设置和zkServer设置的心跳时间有关系
int SESSION_TIMEOUT=4000; // 定义用于保存rpc通信服务端的地址信息的目录
String REGISTRY_PATH="/rpc"; // 定义数据存放的具体目录
String DATA_PATH=REGISTRY_PATH+"/data";
}嫁衣,张丕林,井冈山天气
package cn.wolfcode.rpc.register;import lombok.AllArgsConstructor;import lombok.Getter;import lombok.NoArgsConstructor;import lombok.Setter;import org.apache.zookeeper.*;import org.apache.zookeeper.data.Stat;import org.slf4j.Logger;import org.slf4j.LoggerFactory;@Setter@Getter@AllArgsConstructor()@NoArgsConstructorpublic class RpcRegistry { public static final Logger LOGGER=LoggerFactory.getLogger(RpcRegistry.class); //zkServer的地址信息
private String registryAddress; //zk客户端程序
private ZooKeeper zooKeeper; public void createNode(String data) throws Exception{ //创建一个客户端程序, 对于注册可以不用监听事件
zooKeeper= new ZooKeeper(registryAddress, Constant.SESSION_TIMEOUT, new Watcher() { @Override
public void process(WatchedEvent event) {
}
}); if(zooKeeper!=null){ try{ //判断注册的目录是否存在
Stat stat = zooKeeper.exists(Constant.REGISTRY_PATH, false); if(stat==null){ //如果不存在, 创建一个持久的节点目录
zooKeeper.create(Constant.REGISTRY_PATH,null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
} //创建一个临时的序列节点,并且保存数据信息
zooKeeper.create(Constant.DATA_PATH,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
}catch (Exception e){
LOGGER.error("",e);
e.printStackTrace();
}
}else{
LOGGER.debug("zooKeeper connect is null");
}
} //测试程序
public static void main(String[] args) throws Exception {
RpcRegistry rp唐唯唯cRegistry = new RpcRegistry();
rpcRegistry.setRegistryAddress("192.168.158.151:2181");
rpcRegistry.createNode("testdata"); //让程序等待输入,程序一直处于运行状态
System.in.read();
}
}
package cn.wolfcode.rpc.register;import lombok.Getter;import lombok.Setter;import org.apache.zookeeper.WatchedEvent;import org.apache.zookeeper.Watcher;import org.apache.zookeeper.ZooKeeper;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.util.ArrayList;import java.util.List;import java.util.Random;@Setter@Getter//地址发现,用于实时的获取最新的RPC服务信息public class RpcDiscover { public static final Logger LOGGER=LoggerFactory.getLogger(RpcRegistry.class); //服务端地址 zkServer的地址
private String registryAddress; //获取到的所有提供服务的服务器列表
private volatile List dataList=new ArrayList<>(); private ZooKeeper zooKeeper=null; //初始化zkClient客户端
public RpcDiscover(String registryAddress) throws Exception { this.registryAddress = registryAddress;
zooKee金珍锡per=new ZooKeeper(registryAddress, Constant.SESSION_TIMEOUT, new Watcher() { @Override
public void process(WatchedEvent watchedEvent) { if(watchedEvent.getType()==Event.EventType.NodeChildrenChanged){ //监听zkServer的服务器列表变化
watchNode();
}
}
}); //获取节点相关数据
watchNode();
} // 从dataList列表随机获取一个可用的服务端的地址信息给rpc-client
public String discover(){ int size=dataList.size(); if(size>0){ int index= new Random().nextInt(size); return dataList.get(index);
} throw new RuntimeException("没有找到对应的服务器");
} //监听服务端的列表信息
private void watchNode(){ try{ //获取子节点信息
List nodeList = zooKeeper.getChildren(Constant.REGISTRY_PATH, true);
List dataList=new ArrayList<>(); for (String node : nodeList) { byte[] bytes = zooKeeper.getData(Constant.REGISTRY_PATH + "/" + node, false, null);
dataList.add(new String(bytes));
} this.dataList=dataList;
}catch (Exception e){
LOGGER.error("",e);
e.printStackTrace();
}
} //测试程序
public static void main(String[] args) throws Exception { //打印获取到的连接地址信息
System.out.println(new RpcDiscover("192.168.158.151:218许美静酒店事件1").discover());
System.in.re闽jad();
}
}
  • Constant常量定义,设置连接ZKServer的相关参数
  • RpcRegistry:往注册中心ZKServer设置地址信息,RPC-Server需要使用
  • RpcDiscover: 从注册中心ZKServer获取服务端的网络地址信息 RPC-client需要使用
  1. rpc-common
  2. 定义RPC通信的请求消息和响应消息的规则,以及消息的序列化和反序列化的帮助类,主要包括
  3. 具体代码如下
package cn.wolfcode.rpc.common;import lombok.*;@Setter@Getter@NoArgsConstructor@AllArgsConstructor@ToString//RPC通信的数据请求规则public class RpcRequest { // 请求消息的消息Id
private String requestId; // 请求的具体的类名(接口名称)
private String className; // 请求的具体的方法名称
private String methodName; // 请求的方法参数类型列表
private Class
private Object[] parameters;
}
package cn.wolfcode.rpc.common;import lombok.*;@Setter@Getter@NoArgsConstructor@AllArgsConstructor@ToString//RPC通信消息的响应数据规则public class RpcResponse { //响应的消息id
private String responseId; //请求的消息id
private String requestId; // 响应的消息是否成功
private boolean success; // 响应的数据结果
private Object result; // 如果有异常信息,在该对象中记录异常信息
private Throwable throwable;
}
package cn.wolfcode.rpc.common;import com.dyuproject.protostuff.LinkedBuffer;import com.dyuproject.protostuff.ProtostuffIOUtil;import com.dyuproject.protostuff.Schema;import com.dyuproject.protostuff.runtime.RuntimeSchema;import org.objenesis.Objenesis;import org.objenesis.ObjenesisStd;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;/**
* 序列化工具类(基于 Protostuff 实现) 用于把对象序列化字节数组, 把字节数组反序列化对象
*/public class SerializationUtil { private static Map
} /**
* 获取类的schema
*南京杜爱欣 @param cls
* @return
*/
@S舒画苏文昊uppressWarnings("unchecked") private static Schema getSchema(Class cls) {
Schema schema = (Schema) cachedSchema.get(cls); if (schema == null) {
schema = RuntimeSchema.createFrom(cls); if (schema != null) {
cachedSchema.put(cls, schema);
}
} return schema;
} /**
* 序列化(对象 -> 字节数组)
*/
@SuppressWarnings("unchecked") public static byte[] serialize(T obj) {
Class cls = (Class) obj.getClass();
LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE); try {
Schema schema = getSchema(cls); return ProtostuffIOUtil.toByteArray(obj, schema, buffer);//序列化
} catch (Exception e) { throw new IllegalStateException(e.getMessage(), e);
} finally {
buffer.clear();
}
} /**
* 反序列化(字节数组 -> 对象)
*/
public static T deserialize(byte[] data, Class cls) { try { /*
* 如果一个类没有参数为空的构造方法时候,那么你直接调用newInstance方法试图得到一个实例对象的时候是会抛出异常的
* 通过ObjenesisStd可以完美的避开这个问题
* */
T message = (T) objenesis.newInstance(cls);//实例化
Schema schema = getSchema(cls);//获取类的schema
ProtostuffIOUtil.mergeFrom(data, message, schema); return message;
} catch (Exception e) { throw new IllegalStateException(e.getMessage(), e);
}
}
}
package cn.wolfcode.rpc.common;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.MessageToByteEncoder;//对传递的消息进行编码, 因为是请求/响应对象的传递,先编码为字节数组在发送到服务器解码public class RpcEncoder extends MessageToByteEncoder { // 传递的数据的对象类型
private Class genericClass; public RpcEncoder(Class genericClass) { this.genericClass = genericClass;
} @Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { if(genericClass.i徐纪罡sInstance(msg)){ //序列化请求消息为字节数组
byte[] bytes = SerializationUtil.serialize(msg); // 把数据写入到下一个通道(channel)或者是发往服务端
out.writeBytes(bytes);
}
}
}
package cn.wolfcode.rpc.common;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.ByteToMessageDecoder;import java.util.List;//对传递的消息进行解码, 接受到的数据是字节数组,需要把数组转换为对应的请求/响应消息对象public class RpcDecoder extends ByteToMessageDecoder { private Class
} @Override
//解码方法,把字节数组转换为消息对象
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List

out) throws Exception { //消息的长度
int size=in.readableBytes(); if(size<4){//保证所有的消息都完全接受完成
return;
} byte[] bytes =new byte[size]; //把传递的字节数组读取到bytes中
in.readBytes(bytes); // 反序列化为对象(RPCRequest/RPCResponse对象)
Object object = SerializationUtil.deserialize(bytes, genericClass); //输出对象
out.add(object); //刷新缓存
ctx.flush();
}
}

  • RpcRequest 请求消息封装对象
  • RpcResponse 响应消息封装对象
  • SerializationUtil 消息的序列化,烦序列化帮助类
  • RpcEncoder 把消息对象转换为字节数组进行通信
  • RpcDecoder 把获取到的字节数组转换为对应的消息对象
  1. rpc-server
  2. RPC服务端,启动RPC服务,扫描app-server中的所有可以提供的服务列表并保存,接受RPC客户端的消息并且通过反射调用具体的方法,响应RPC客户端,把方法执行结果返回到RPC客户端
  3. 主要包括:
package cn.wolfcode.rpc.server;import java.lang.annotation.ElementType;import java.lang.annotation.Retention;import java.lang.annotation.RetentionPolicy;import java.lang.annotation.Target;/**
* 这个注解用于贴在每个提供服务的实现类,
* 在Spring容器启动的时候,自动扫描到贴了该注解的所有的服务
*/@Retention(RetentionPolicy.RUNTIME)@Target({ElementType.TYPE})public @interface RpcService { public Class
}
package cn.wolfcode.rpc.server;import cn.wolfcode.rpc.common.RpcDecoder;import cn.wolfcode.rpc.common.RpcEncoder;import cn.wolfcode.rpc.common.RpcRequest;import cn.wolfcode.rpc.common.RpcResponse;import cn.wolfcode.rpc.register.RpcRegistry;import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChwdgafannelOption;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import lombok.AllArgsConstructor;import lombok.Getter;import lombok.NoArgsConstructor;import lombok.Setter;import org.apache.commons.collections4.MapUtils;import org.springframework.beans.BeansException;import org.springframework.beans.factory.InitializingBean;import org.springframework.context.ApplicationContext;import org.springframework.context.ApplicationContextAware;import java.util.HashMap;import java.util.Map;@Setter@Getter@NoArgsConstructor@AllArgsConstructor//RPC服务端启动,实现Spring的感知接口public class RpcServer implements ApplicationContextAware,InitializingBean { //用于保存所有提供服务的方法, 其中key为类的全路径名, value是所有的实现类
private final Map serviceBeanMap=new HashMap<>(); //rpcRegistry 用于注册相关的地址信息
private RpcRegistry rpcRegistry; //提供服务的地址信息 格式为 192.168.158.151:9000 类似
private String serverAddress; //在Spring容器启动完成后会执行该方法
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { //获取到所有贴了RpcService注解的Bean对象
Map serviceBeanMap = applicationContext.getBeansWithAnnotation(RpcService.class); if(MapUtils.isNotEmpty(serviceBeanMap)){ for (Object object : serviceBgivemefive什么意思eanMap.values()) { //获取到类的路径名称
String serviceName = object.getClass().getAnnotation(RpcService.class).value().getName(); //把获取到的信息保存到serviceBeanMap中
this.serviceBeanMap.put(serviceName,object);
}
}
System.out.println("服务器: "+serverAddress +" 提供的服务列表: "+ serviceBeanMap );
} // 初始化完成后执行
@Override
public void afterPropertiesSet(张云成) throws Exception { //创建服务端的通信对象
ServerBootstrap server = new ServerBootstrap(); // 创建异步通信的事件组 用于建立TCP连接的
NioEventLoopGroup bossGroup = new NioEventLoopGroup(); // 创建异步通信的事件组 用于处理Channel(通道)的I/O事件
NioEventLoopGroup workerGroup = new NioEventLoopGroup(); try{ //开始设置server的相关参数
server.group(bossGroup,workerGroup) //启动异步ServerSocket
.channel(NioServerSocketChannel.class) //初始化通道信息
.childHandler(new ChannelInitializer() { @Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new RpcDecoder(RpcRequest.class))//1 解码请求参数
.addLast(new RpcEncoder(RpcResponse.class))//2 编码响应信息
.addLast(new RpcServerHandler(serviceBeanMap));//3 请求处理
}
}).option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);;
String host=serverAddress.split(":")[0] ;//获取到主机地址
int port=Integer.valueOf(serverAddress.split(":")[1]);//端口
ChannelFuture future = server.bind(host, port).sync();//开启异步通信服务
System.out.println("服务器启动成功:"+future.channel().localAddress());
rpcRegistry.createNode(serverAddress);
System.out.println("向zkServer注册服务地址信息");
future.channel().closeFuture().sync();//等待通信完成
}catch (Exception e){
e.printStackTrace();
}finally { //优雅的关闭socket
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
package cn.wolfcode.rpc.server;import cn.wolfcode.rpc.common.RpcRequest;import cn.wolfcode.rpc.common.RpcResponse;import io.netty.channel.Ch柯里思annelFutureListener;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import lombok.Getter;import lombok.NoArgsConstructor;import lombok.Setter;import java.lang.reflect.Method;import java.util.Map;import java.uti欲望乐园l.UUID;@Setter@Getter@NoArgsConstructorpublic class RpcServerHandler extends ChannelInboundHandlerAdapter{ private Map serviceBeanMap; public RpcServerHandler(Map serviceBeanMap) { this.serviceBeanMap = serviceBeanMap;
} @Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("RpcServerHandler.channelRead");
System.out.println(msg);
RpcRequest rpcRequest= (RpcRequest) msg;
RpcResponse rpcResponse=handler(rpcRequest); //告诉客户端,关闭socket连接
ctx.writeAndFlush(rpcResponse).addListener(ChannelFutureListener.CLOSE);
} private RpcResponse handler(RpcRequest rpcRequest) { //创建一个响应消息对象
RpcResponse rpcResponse =new RpcResponse(); //设置响应消息ID
rpcResponse.setResponseId(UUID.randomUUID().toString()); //请求消息ID
rpcResponse.setRequestId(rpcRequest.getRequestId()); try{ //获取到类名(接口名称)
String className = rpcRequest.getClassName(); //获取到方法名
String methodName = rpcRequest.getMethodName(); //获取到参数类型列表
Class
Object[] parameters = rpcRequest.getParameters(); //获取到具字节码对象
Class
Object serviceBean = serviceBeanMap.get(className); if(serviceBean==null){ throw new RuntimeException(className+"没有找到对应的serviceBean:"+className+":beanMap:"+serviceBeanMap);
} //反射调用方法
Method method = clz.getMethod(methodName, parameterTypes); if(method==null) throw new RuntimeException("没有找到对应的方法");
Object result = method.invoke(serviceBean, parameters);
rpcResponse.setSuccess(true); //设置方法调用的结果
rpcResponse.setResult(result);
}catch (Exception e){
rpcResponse.setSuccess(false);
rpcResponse.setThrowable(e);
e.printStackTrace();
} return rpcResponse;
}
}
  • RpcService 定义一个注解,用于标记服务程序的提供者,通过Spring扫描出所有的服务并且保存
  • RpcServerHandler 处理RPC客户端请求,调用服务提供者的具体方法,响应执行结果
  • RpcServer 扫描所有的服务(标记了@RPCService的类),启动RPC服务
  1. rpc-client
  2. RPC客户端,通过网络通信往RPC服务端发送请求调用消息,接受服务端的响应消息,配置动态代理类,所有的方法调用都通过网络调用发送到RPC服务端
  3. 其中包括的主要代码:
  4. 具体实现代码:
package cn.wolfcode.rpc.client;import cn.wolfcode.rpc.common.RpcDecoder;import cn.wolfcode.rpc.common.RpcEncoder;import cn.wolfcode.rpc.common.RpcRequest;import cn.wolfcode.rpc.common.RpcResponse;import cn.wolfcode.rpc.register.RpcDiscover;import io.netty.bootstrap.Bootstrap;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;//RPC通信客户端,往服务端发送请求,并且接受服务端的响应public class RpcClient extends SimpleChannelInboundHandler { //消息响应对象
private RpcResponse rpcResponse; //消息请求对象
private RpcRequest rpcRequest; // 同步锁 资源对象
private Object object=new Object(); // 用于获取服务地址列表信息
private RpcDiscover rpcDiscover; //构造函数
public RpcClient(RpcRequest rpcRequest,RpcDiscover rpcDiscover) { this.rpcDiscover = rpcDiscover; this.rpcRequest=rpcRequest;
} @Override
protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception { this.rpcResponse=msg;//响应消息
synchronized (object){
ctx.flush();//刷新缓存
object.notifyAll();//唤醒等待
}
} //发送消息
public RpcResponse send() throws Exception { //创建一个socket通信对象
Bootstrap client = new Bootstrap(); //创建一个通信组,负责Channel(通道)的I/O事件的处理
NioEventLoopGroup loopGroup = new NioEventLoopGroup(); try{
client.group(loopGroup)//设置参数
.channel(NioSocketChannel.class)//使用异步socket通信
.handler(new ChannelInitializer() { @Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new RpcEncoder(RpcRequest.class))//编码请求对象
.addLast(new RpcDecoder(RpcResponse.class))//解码响应对象
.addLast(RpcClient.this);//发送请求对象
}
}).option(ChannelOption.SO_KEEPALIVE, true);;
String serverAddress = rpcDiscover.discover();//获取一个服务器地址
String host=serverAddress.split(":")[0]; int port=Integer.valueOf(serverAddress.split(":")[1]);
ChannelFuture future = client.connect(host,port).sync();
System.out.println("客户端准备发送数据:"+rpcRequest);
future.channel().writeAndFlush(rpcRequest).sync(); synchronized (object){
object.wait();//线程等待,等待客户端响应
} if (rpcResponse != null) {
future.channel().closeFuture().sync();//等待服务端关闭socket
} return rpcResponse;
}finally {
loopGroup.shutdownGracefully();//优雅关闭socket
}
} /**
* 异常处理
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
ctx.close();
}
}
pac眼镜蛇11燃烧汽车kage cn.wolfcode.rpc.client;import cn.wolfcode.rpc.common.RpcRequest;import cn.wolfcode.rpc.common.RpcResponse;import cn.wolfcode.rpc.register.RpcDiscover;import lombok.Getter;import lombok.Setter;import java.lang.reflect.InvocationHandler;import java.lang.reflect.Method;import java.lang.reflect苏燃陆廷风.Proxy;import java.util.UUID;@Setter@Getter//动态代理类,用于获取到每个类的代理对象//对于被代理对象的所有的方法调用都会执行invoke方法public class RpcProxy { //用于获取到RPC-Server的地址信息
private RpcDiscover rpcDiscover; @SuppressWarnings("all") public T getInstance(Class interfaceClass){
T instance = (T)Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { //创建请求对象
RpcRequest rpcRequest = new RpcRequest(); //获取到被调用的类名 和RPC-Server中的serviceMap中的key进行匹配
String className=method.getDeclaringClass().getName(); //获取到方法的参数列表
Class
rpcRequest.setRequestId(UUID.randomUUID().toString());
rpcRequest.setClassName(className);//类名
rpcRequest.setParameterTypes(parameterTypes);//参数类型列表
rpcRequest.setParameters(args);//参数列表
rpcRequest.setMethodName(method.getName());//调用的放方法名称
RpcResponse rpcResponse = new RpcClient(rpcRequest, rpcDiscover).send();//创建一个RPCclient对象,并且发送消息到服务端
//返回调用结果
return rpcResponse.getResult();
}
}); //返回一个代理对象
return instance;
}
}
  • RpcProxy 对于每一个类都创建一个动态代理对象,并且在invoke方法创建rpc客户端并且发送网络通信请求
  • RpcClient RPC通信客户端,启动RPC通信服务,创建TCP连接,发送请求,接受响应
  1. app-common
  2. 这是具体应用的通用模块,和具体的项目结构有关系,这里主要包括接口定义和JavaBean对象的定义
  3. 具体代码为:
package cn.wolfcode.app.common;public interface IProductService { /**
* 保存产品
* @param product
*/
void save(Product product); /**
* 根据产品id删除产品
* @param productId
*/
void deleteById(Long productId); /**
* 修改产品信息
* @param product
*/
void update(Product product); /**
* 根据产品id获取到产品信息
* @param productId
* @return
*/
Product get(Long productId);
}
pac禾博士为什么那么便宜kage cn.wolfcode.app.common;import lombok.*;import java.math.BigDecimal;/**
* 产品信息
*/@Setter@Getter@ToString@AllArgsConstructor@NoArgsConstructorpublic class Product { private Long id;//id
private String sn;//产品编号
private String name;//产品名称
private BigDecimal price;//产品价格}
  1. app-server
  2. 这个模块主要是定义服务的具体实现和启动Spring容器,在启动Spring容器的时候需要创建RpcRegistry,RpcServer对象
  3. 具体代码实现:
package cn.wolfcode.app.server;import cn.wolfcode.app.common.IProductService;import cn.wolfcode.app.common.Product;import cn.wolfcode.rpc.server.RpcService;import org.springframework.stereotype.Component;import java.math.BigDecimal;@Component@RpcService(IProductService.class)public class ProductServiceImpl implements IProductService { @Override
public void save(Product product) {
System.out.println("产品保存成功: "+product);
} @Override
public void deleteById(Long productId) {
System.out.println("产品删除成功: "+ productId);
} @Override
public void update(Product product) {
System.out.println("产品修改成功: "+ product);
} @Override
public Product get(Long productId) {
System.out.println("产品获取成功"); return new Product(1L,"001","笔记本电脑",BigDecimal.TEN);
}
}
package cn.wolfcode.app.server;import org.springframework.context.support.ClassPathXmlApplicationContext;public class BootAppServer { public static void main(String[] args) { //启动Spring容器
new ClassPathXmlApplicationContext("classpath:application.xml");
}
}

其中配置文件:


xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/sp刘世宇哪里人ring-context.xsd">









# zookeeper serverregistry.address=192.168.158.151:2181# rpc serverserver.address=192.168.158.1:9090
  • application.xml Spring的配置文件
  • log4j.properties 日志配置文件
  • rpc.properties 服务提供者的地址和端口 以及zkServer的连接地址和端口
  1. app-client
  2. 通过Spring的配置创建RpcDiscover对象和RpcProxy对象,其中RpcDiscover用于从注册中心获取到服务的地址信息,RpcProxy用于创建类的动态代理对象
  3. 测试类:使用Spring的Junit进行测试
package cn.wolfcode.app.client;import cn.wolfcode.app.common.IProductService;impor邹瀚枢t cn.wolfcode.app.common.Product;import cn.wolfcode.rpc.client.RpcProxy;import org.junit.Before;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.test.context.ContextConfiguration;import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;import java.math.BigDecimal;//模拟客户端启动@RunWith(SpringJUnit4ClassRunner.class)@ContextConfiguration(locations="classpath:application.xml")publi被吃奶c class APP { @Autowired
private RpcProxy rpcProxy; private IProductService productService; @Before
public void init() {
productService = rpcProxy.getInstance(IProductService.class);
} @Test
public void testSave() throws Exception {
productService.save(new Product(2L,"002","内衣",BigDecimal.TEN));
} @Test
public void testDelete() throws Exception {
productService.deleteById(2L);
} @Test
public void testUpdate() throws Exception {
productService.update(new Product(2L,"002","内衣",BigDecimal.ONE));
} @Test
public void testGet() throws Exception {
Product product = productService.get(1L);
System.out.println("获取到的产品信息为:"+product);
}
}

配置文件信息

application.xml


xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd">








log4j.properties

log4j.rootLogger=ERROR,consolelog4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%m%n
log4j.logger.cn.wolfcode.rpc=DEBUG

rpc.properties

# zookeeper serverregistry.address=192.168.158.151:2181

对于本文的完整代码下载地址为 https://gitee.com/heshengjun/rpcdemo.git

如果要正常运行,请部署一个zookeeper注册中心,修改rpc.properites的地址即可

  • 先运行app-server中的BootAppServer
  • 在运行app-client中的APP测试用例
版权声明

本文仅代表作者观点,不代表本站立场。
本文系作者授权发表,未经许可,不得转载。