100 行代码透彻解析 RPC 原理!
程序员的成长之路
共 18429字,需浏览 37分钟
· 2022-05-19
阅读本文大概需要 12 分钟。
来自:https://sourl.cn/HpZHvy
引 言
一次基本的 RPC 调用会涉及到什么?
基本实现
/**
* @author 孙浩
* @Descrption 服务接口
***/
public interface HelloService {
String sayHello(String somebody);
}
/**
* @author 孙浩
* @Descrption 服务实现
***/
public class HelloServiceImpl implements HelloService {
@Override
public String sayHello(String somebody) {
return "hello " + somebody + "!";
}
}
<xsd:element name="service">
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:attribute name="interface" type="xsd:string" use="required"/>
<xsd:attribute name="timeout" type="xsd:int" use="required"/>
<xsd:attribute name="serverPort" type="xsd:int" use="required"/>
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="weight" type="xsd:int" use="optional"/>
<xsd:attribute name="workerThreads" type="xsd:int" use="optional"/>
<xsd:attribute name="appKey" type="xsd:string" use="required"/>
<xsd:attribute name="groupName" type="xsd:string" use="optional"/>
xsd:extension>
xsd:complexContent>
xsd:complexType>
xsd:element>
http\://www.storm.com/schema/storm-service.xsd=META-INF/storm-service.xsd
http\://www.storm.com/schema/storm-reference.xsd=META-INF/storm-reference.xsd
http\://www.storm.com/schema/storm-service=com.hsunfkqm.storm.framework.spring.StormServiceNamespaceHandler
http\://www.storm.com/schema/storm-reference=com.hsunfkqm.storm.framework.spring.StormRemoteReferenceNamespaceHandler
<bean id="helloService" class="com.hsunfkqm.storm.framework.test.HelloServiceImpl"/>
<storm:service id="helloServiceRegister"
interface="com.hsunfkqm.storm.framework.test.HelloService"
ref="helloService"
groupName="default"
weight="2"
appKey="ares"
workerThreads="100"
serverPort="8081"
timeout="600"/>
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
/**
* @author 孙浩
* @Descrption 服务发布自定义标签
***/
public class StormServiceNamespaceHandler extends NamespaceHandlerSupport {
@Override
public void init() {
registerBeanDefinitionParser("service", new ProviderFactoryBeanDefinitionParser());
}
}
protected Class getBeanClass(Element element) {
return ProviderFactoryBean.class;
}
protected void doParse(Element element, BeanDefinitionBuilder bean) {
try {
String serviceItf = element.getAttribute("interface");
String serverPort = element.getAttribute("serverPort");
String ref = element.getAttribute("ref");
// ....
bean.addPropertyValue("serverPort", Integer.parseInt(serverPort));
bean.addPropertyValue("serviceItf", Class.forName(serviceItf));
bean.addPropertyReference("serviceObject", ref);
//...
if (NumberUtils.isNumber(weight)) {
bean.addPropertyValue("weight", Integer.parseInt(weight));
}
//...
} catch (Exception e) {
// ...
}
}
/**
* @author 孙浩
* @Descrption 服务发布
***/
public class ProviderFactoryBean implements FactoryBean, InitializingBean {
//服务接口
private Class> serviceItf;
//服务实现
private Object serviceObject;
//服务端口
private String serverPort;
//服务超时时间
private long timeout;
//服务代理对象,暂时没有用到
private Object serviceProxyObject;
//服务提供者唯一标识
private String appKey;
//服务分组组名
private String groupName = "default";
//服务提供者权重,默认为 1 , 范围为 [1-100]
private int weight = 1;
//服务端线程数,默认 10 个线程
private int workerThreads = 10;
@Override
public Object getObject() throws Exception {
return serviceProxyObject;
}
@Override
public Class> getObjectType() {
return serviceItf;
}
@Override
public void afterPropertiesSet() throws Exception {
//启动 Netty 服务端
NettyServer.singleton().start(Integer.parseInt(serverPort));
//注册到 zk, 元数据注册中心
ListproviderServiceList = buildProviderServiceInfos();
IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton();
registerCenter4Provider.registerProvider(providerServiceList);
}
}
//================RegisterCenter#registerProvider======================
@Override
public void registerProvider(final ListserviceMetaData) {
if (CollectionUtils.isEmpty(serviceMetaData)) {
return;
}
//连接 zk, 注册服务
synchronized (RegisterCenter.class) {
for (ProviderService provider : serviceMetaData) {
String serviceItfKey = provider.getServiceItf().getName();
Listproviders = providerServiceMap.get(serviceItfKey);
if (providers == null) {
providers = Lists.newArrayList();
}
providers.add(provider);
providerServiceMap.put(serviceItfKey, providers);
}
if (zkClient == null) {
zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
}
//创建 ZK 命名空间/当前部署应用 APP 命名空间/
String APP_KEY = serviceMetaData.get(0).getAppKey();
String ZK_PATH = ROOT_PATH + "/" + APP_KEY;
boolean exist = zkClient.exists(ZK_PATH);
if (!exist) {
zkClient.createPersistent(ZK_PATH, true);
}
for (Map.Entry> entry : providerServiceMap.entrySet()) {
//服务分组
String groupName = entry.getValue().get(0).getGroupName();
//创建服务提供者
String serviceNode = entry.getKey();
String servicePath = ZK_PATH + "/" + groupName + "/" + serviceNode + "/" + PROVIDER_TYPE;
exist = zkClient.exists(servicePath);
if (!exist) {
zkClient.createPersistent(servicePath, true);
}
//创建当前服务器节点
int serverPort = entry.getValue().get(0).getServerPort();//服务端口
int weight = entry.getValue().get(0).getWeight();//服务权重
int workerThreads = entry.getValue().get(0).getWorkerThreads();//服务工作线程
String localIp = IPHelper.localIp();
String currentServiceIpNode = servicePath + "/" + localIp + "|" + serverPort + "|" + weight + "|" + workerThreads + "|" + groupName;
exist = zkClient.exists(currentServiceIpNode);
if (!exist) {
//注意,这里创建的是临时节点
zkClient.createEphemeral(currentServiceIpNode);
}
//监听注册服务的变化,同时更新数据到本地缓存
zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
@Override
public void handleChildChange(String parentPath, ListcurrentChilds) throws Exception {
if (currentChilds == null) {
currentChilds = Lists.newArrayList();
}
//存活的服务 IP 列表
ListactivityServiceIpList = Lists.newArrayList(Lists.transform(currentChilds, new Function () {
@Override
public String apply(String input) {
return StringUtils.split(input, "|")[0];
}
}));
refreshActivityService(activityServiceIpList);
}
});
}
}
}
public void start(final int port) {
synchronized (NettyServer.class) {
if (bossGroup != null || workerGroup != null) {
return;
}
bossGroup = new NioEventLoopGroup();
workerGroup = new NioEventLoopGroup();
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap
.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childOption(ChannelOption.TCP_NODELAY, true)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//注册解码器 NettyDecoderHandler
ch.pipeline().addLast(new NettyDecoderHandler(StormRequest.class, serializeType));
//注册编码器 NettyEncoderHandler
ch.pipeline().addLast(new NettyEncoderHandler(serializeType));
//注册服务端业务逻辑处理器 NettyServerInvokeHandler
ch.pipeline().addLast(new NettyServerInvokeHandler());
}
});
try {
channel = serverBootstrap.bind(port).sync().channel();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, StormRequest request) throws Exception {
if (ctx.channel().isWritable()) {
//从服务调用对象里获取服务提供者信息
ProviderService metaDataModel = request.getProviderService();
long consumeTimeOut = request.getInvokeTimeout();
final String methodName = request.getInvokedMethodName();
//根据方法名称定位到具体某一个服务提供者
String serviceKey = metaDataModel.getServiceItf().getName();
//获取限流工具类
int workerThread = metaDataModel.getWorkerThreads();
Semaphore semaphore = serviceKeySemaphoreMap.get(serviceKey);
if (semaphore == null) {
synchronized (serviceKeySemaphoreMap) {
semaphore = serviceKeySemaphoreMap.get(serviceKey);
if (semaphore == null) {
semaphore = new Semaphore(workerThread);
serviceKeySemaphoreMap.put(serviceKey, semaphore);
}
}
}
//获取注册中心服务
IRegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton();
ListlocalProviderCaches = registerCenter4Provider.getProviderServiceMap().get(serviceKey);
Object result = null;
boolean acquire = false;
try {
ProviderService localProviderCache = Collections2.filter(localProviderCaches, new Predicate() {
@Override
public boolean apply(ProviderService input) {
return StringUtils.equals(input.getServiceMethod().getName(), methodName);
}
}).iterator().next();
Object serviceObject = localProviderCache.getServiceObject();
//利用反射发起服务调用
Method method = localProviderCache.getServiceMethod();
//利用 semaphore 实现限流
acquire = semaphore.tryAcquire(consumeTimeOut, TimeUnit.MILLISECONDS);
if (acquire) {
result = method.invoke(serviceObject, request.getArgs());
//System.out.println("---------------"+result);
}
} catch (Exception e) {
System.out.println(JSON.toJSONString(localProviderCaches) + " " + methodName+" "+e.getMessage());
result = e;
} finally {
if (acquire) {
semaphore.release();
}
}
//根据服务调用结果组装调用返回对象
StormResponse response = new StormResponse();
response.setInvokeTimeout(consumeTimeOut);
response.setUniqueKey(request.getUniqueKey());
response.setResult(result);
//将服务调用返回对象回写到消费端
ctx.writeAndFlush(response);
} else {
logger.error("------------channel closed!---------------");
}
}
/**
* @author 孙浩
* @Descrption
***/
public class StormRequest implements Serializable {
private static final long serialVersionUID = -5196465012408804755L;
//UUID, 唯一标识一次返回值
private String uniqueKey;
//服务提供者信息
private ProviderService providerService;
//调用的方法名称
private String invokedMethodName;
//传递参数
private Object[] args;
//消费端应用名
private String appName;
//消费请求超时时长
private long invokeTimeout;
// getter/setter
}
/**
* @author 孙浩
* @Descrption
***/
public class StormResponse implements Serializable {
private static final long serialVersionUID = 5785265307118147202L;
//UUID, 唯一标识一次返回值
private String uniqueKey;
//客户端指定的服务超时时间
private long invokeTimeout;
//接口调用返回的结果对象
private Object result;
//getter/setter
}
public Object getProxy() {
return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class>[]{targetInterface}, this);
}
//服务接口名称
String serviceKey = targetInterface.getName();
//获取某个接口的服务提供者列表
IRegisterCenter4Invoker registerCenter4Consumer = RegisterCenter.singleton();
ListproviderServices = registerCenter4Consumer.getServiceMetaDataMap4Consume().get(serviceKey);
//根据软负载策略,从服务提供者列表选取本次调用的服务提供者
ClusterStrategy clusterStrategyService = ClusterEngine.queryClusterStrategy(clusterStrategy);
ProviderService providerService = clusterStrategyService.select(providerServices);
/**
* @author 孙浩
* @Descrption Netty 消费端 bean 代理工厂
***/
public class RevokerProxyBeanFactory implements InvocationHandler {
private ExecutorService fixedThreadPool = null;
//服务接口
private Class> targetInterface;
//超时时间
private int consumeTimeout;
//调用者线程数
private static int threadWorkerNumber = 10;
//负载均衡策略
private String clusterStrategy;
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
...
//复制一份服务提供者信息
ProviderService newProvider = providerService.copy();
//设置本次调用服务的方法以及接口
newProvider.setServiceMethod(method);
newProvider.setServiceItf(targetInterface);
//声明调用 AresRequest 对象,AresRequest 表示发起一次调用所包含的信息
final StormRequest request = new StormRequest();
//设置本次调用的唯一标识
request.setUniqueKey(UUID.randomUUID().toString() + "-" + Thread.currentThread().getId());
//设置本次调用的服务提供者信息
request.setProviderService(newProvider);
//设置本次调用的方法名称
request.setInvokedMethodName(method.getName());
//设置本次调用的方法参数信息
request.setArgs(args);
try {
//构建用来发起调用的线程池
if (fixedThreadPool == null) {
synchronized (RevokerProxyBeanFactory.class) {
if (null == fixedThreadPool) {
fixedThreadPool = Executors.newFixedThreadPool(threadWorkerNumber);
}
}
}
//根据服务提供者的 ip,port, 构建 InetSocketAddress 对象,标识服务提供者地址
String serverIp = request.getProviderService().getServerIp();
int serverPort = request.getProviderService().getServerPort();
InetSocketAddress inetSocketAddress = new InetSocketAddress(serverIp, serverPort);
//提交本次调用信息到线程池 fixedThreadPool, 发起调用
FutureresponseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request));
//获取调用的返回结果
StormResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS);
if (response != null) {
return response.getResult();
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return null;
}
// ...
}
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, StormResponse response) throws Exception {
//将 Netty 异步返回的结果存入阻塞队列,以便调用端同步获取
RevokerResponseHolder.putResultValue(response);
}
//提交本次调用信息到线程池 fixedThreadPool, 发起调用
FutureresponseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request));
//获取调用的返回结果
StormResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS);
if (response != null) {
return response.getResult();
}
//===================================================
//从返回结果容器中获取返回结果,同时设置等待超时时间为 invokeTimeout
long invokeTimeout = request.getInvokeTimeout();
StormResponse response = RevokerResponseHolder.getValue(request.getUniqueKey(), invokeTimeout);
测 试
/**
* @author 孙浩
* @Descrption
***/
public class MainServer {
public static void main(String[] args) throws Exception {
//发布服务
final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("storm-server.xml");
System.out.println(" 服务发布完成");
}
}
public class Client {
private static final Logger logger = LoggerFactory.getLogger(Client.class);
public static void main(String[] args) throws Exception {
final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("storm-client.xml");
final HelloService helloService = (HelloService) context.getBean("helloService");
String result = helloService.sayHello("World");
System.out.println(result);
for (;;) {
}
}
}
结 果
总 结
加载服务接口,并缓存
服务注册,将服务接口以及服务主机信息写入注册中心(本例使用的是 zookeeper)
启动网络服务器并监听
反射,本地调用
代理服务接口生成代理对象
服务发现(连接 zookeeper,拿到服务地址列表,通过客户端负载策略获取合适的服务地址)
远程方法调用(本例通过 Netty,发送消息,并获取响应结果)
推荐阅读:
内容包含Java基础、JavaWeb、MySQL性能优化、JVM、锁、百万并发、消息队列、高性能缓存、反射、Spring全家桶原理、微服务、Zookeeper......等技术栈!
⬇戳阅读原文领取! 朕已阅
评论
老爸嘲讽我了,写破代码一年就挣十几万,他在工地带50个工人,一个月光人头费就3万,让我滚回去跟他干!
点击上方 "大数据肌肉猿"关注, 星标一起成长点击下方链接,进入高质量学习交流群今日更新| 1052个转型案例分享-大数据交流群来自:网络,侵删有个网友的父亲是做工程的,天天就嘲笑他,说他天天写着破代码有啥用,一年就拿个十多万的死工资,然后告诉他自己在工地里面带了50个工人,一个月能抽三万
程序源代码
0
好未来测开一面,挺简单!(0428面试原题解析)
大家好,我是二哥呀。今天继续给大家分享春招面试题《好未来测开一面原题》,附详细答案,我会用通俗易懂+手绘图的方式,让天下所有的面渣都能逆袭 😁二哥的 Java 面试指南内容较长,建议正在冲刺 24 届春招和 25 届暑期实习、秋招的同学先收藏起来,面试的时候大概率会碰到,1、二哥的 Linux 速查
沉默王二
0
大厂都在用的 Git 代码管理规范 !
👉 欢迎加入小哈的星球 ,你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 / 赠书福利全栈前后端分离博客项目 2.0 版本完结啦, 演示链接:http://116.62.199.48/ ,新项目正在酝酿中
小哈学Java
2
这五款牛逼的 IDEA 插件,堪称代码质量检查利器!
来源:blog.csdn.net/a745233700?type=blog一、Alibaba Java Coding Guidelines二、CheckStyle:三、PMD四、FindBugs:五、SonarLint:总结随着业务的发展,系统会越来越庞大,原本简单稳定的功能,可能在不断迭代后复杂度
码农突围
0
代码 | 使用 MD&A文本测量「企业不确定性感知FEPU」
Tips: 为了更好的阅读体验,建议阅读本文博客版, 链接地址https://textdata.cn/blog/2024-04-25-firm-economic-policy-uncertainty/本文使用的缩写EPU 经济政策不确定性(Economic Policy Uncerta
大邓和他的Python
0
会写代码的总理!全球第一“开源”名门望族
转自:OSC开源社区4 月 15 日,新加坡总理公署发表声明宣布,总理李显龙将于 5 月 15 日辞职,并正式交棒给副总理兼财政部长黄循财。对于李氏家族下一代是否会继续活跃在新加坡政坛,目前外界说法不一。但在开源圈里,李氏家族绝对有一席之地。李显龙有 4 名子女,其本人、次子,以及幼子都有非常专业的
开源前哨
0
原来Matplotlib能画股票K线图!!附代码
之前在一篇文章中提到Matplotlib可视化,甚至可以用来画股票K线图,许多同学也在问代码,这次来发个文回应下。Python用matplotlib绘制K线图,需要配合talib、numpy、mpl_finance等第三方库来使用,效果展示如下:简单讲讲K线图的结构,我不搞股票,所以不太懂,特地查了
Python大数据分析
9
分享一份抓取某东商品名称、价格和评论数的代码
点击上方“Python共享之家”,进行关注回复“资源”即可获赠Python学习资料今日鸡汤芳草已云暮,故人殊未来。大家好,我是皮皮。一、前言前几天在Python白银交流群【邮递员】问了一个Python网络爬虫的问题,提问截图如下:代码如下:import requestsfrom
IT共享之家
0