1. What is RPC
Remote Procedure Call (RPC) is short for Remote Procedure Call. The so-called Remote Procedure Call means that an application synchronously invokes methods running in other computer processes over the network like calling local methods. In other words, RPC is a way of interprocess communication, and its architecture adopts the default C/S architecture. The caller is the client and the service provider is the server. It is mainly used in distributed system service invocation, distributed computing and other scenarios.
2.RPC framework design
As you can see from the above RPC introduction, the core components of RPC are
- Client: service caller.
- Client Stub: stores the address information of the server, packages the request parameters of the Client into network messages, and sends them to the server over the network.
- The Server Stub receives and unpacks the request message sent by the client, and then invokes the local service for processing.
- Server: The real provider of a service.
- Network Service: Underlying transport, either TCP or HTTP.
These components collaborate to complete cross-process service invocations as follows
1. The client invokes the service in the local invocation mode.
2. After receiving the call, the client stub is responsible for assembling methods and parameters into a message body that can be transmitted through the network.
3. The client stub finds the service address and sends the message to the server.
4. The stub server decodes the message after receiving it.
5. The server stub invokes the local service according to the decoding result.
6. The local service executes and returns the result to the server stub.
7. The server stub packages the returned result into a message and sends it to the consumer.
8. The client stub receives the message and decodes it.
9. The client Stub is returned by the server.
10. The client receives the final result.
The goal of the RPC framework is to encapsulate steps 2 through 8 and make these details transparent to the user. To implement the entire process above, based on the Java language, we solved the following technical problems:
1. The serialization
Serialization refers to the above steps: the client Stub and the server Stub need to transfer the data returned by the upper layer to the other process through our network module. For the sake of performance and security, the data needs to be converted into a data format convenient for network transmission. The transfer from application data to network data is serialized, and the transfer from network data to application data is deserialized. Serialization is implemented by default in Java, but given the high concurrency performance requirements of most current applications, I generally use other high-performance serialization implementations such as Protobuf, Kryo, Hessian, Jackson, etc
2. Network transmission
Network transmission we are based on the OSI network layer TCP/UDP socket implementation, but IO processing based on the current high concurrency performance requirements, the traditional blocking IO is not applicable, Java also provides NIO, as well as Java7 NIO.2 implementation
3. Registry
The role of the registry is a RPC service management center, it has two main functions, the first is the service registry, for the RPC server provides service, if you want to know by other call, and call, need to call a method, it unified sent to registry, including addressing ID, into and out and types, etc.; The second is that the registry needs to inform the client of the server-side methods registered with the registry so that the client can make the call. Currently, ZooKeeper is widely used. Zookeeper maintains a service registry and enables multiple service providers to form a cluster. The client obtains the server access address (IP+ port) from the registry to invoke the service
4. Proxy mode
One of the main characteristics of RPC framework is that it encapsulates communication details and makes RPC interface call like calling local methods. To achieve this goal, we need to use proxy mode, and the underlying communication logic is realized by RPC interface proxy class. There are two main approaches to Java proxy implementation, dynamic proxy after Java 5, and Cglib bytecode proxy.
By integrating the above RPC core process and based on Java Spring framework, we can conveniently implement RPC service development, and the detailed implementation process is as follows:
3. Implementation analysis of Dubbo RPC framework
Introduction to the Dubbo framework
Dubbo is an open source RPC communication framework for service governance of Alibaba. In addition to basic RPC capabilities, Dubbo provides service governance capabilities such as service discovery, service degradation, service routing, and cluster fault tolerance. Here we mainly analyze the implementation of RPC function of Dubbo.
Here is Dubbo’s architecture:
- Container Container starts the service provider.
- The service Provider starts and registers the service with Registry.
- The service Consumer starts and subscribes to the service from Registry.
- The service Consumer makes a request to the service Provider.
- Service providers and consumers synchronize monitoring central statistics.
Implementation of Dubbo RPC module
The Dubbo-RPC module contains multiple packages, among which the Dubbo-Rpc-API package defines the interfaces that a specific RPC implementation needs to implement. It provides a unified API layer for custom extensions by clearly defining the interface of RPC implementation. Dubbo’s implementation is highly extensible.
Dubbo supports protocol extensions, which are placed in various packages of the Dubbo-RPC module. For example, the Dubbp-Rpc-Dubbo package contains the Dubbo protocol extension, and the Dubbo-Rpc-HTTP package is an RPC extension implemented by the HTTP protocol. The DUbo-RPC module contains a total of 13 RPC extensions implemented by different protocols
We can choose different RPC implementations based on the requirements of the technology stack. Dubbo is recommended for short message requests. Let’s take an example of how Dubbo abstracts remote procedure calls using the RPC implementation of dubbo protocol, Dubbo-Rpc-Dubbo.
4.Dubbo RPC key source code analysis
The figure above shows the main components that dubbo RPC flows through to complete an RPC method call
1. First, the service Proxy holds an Invoker object. The invoke call is then triggered.
public class InvokerInvocationHandler implements InvocationHandler {
private finalInvoker<? > invoker;public InvokerInvocationHandler(Invoker
handler) {
this.invoker = handler;
}
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {~ ~ ~/** * here is the start of the RPC service invocation ** now invoker is MockClusterInoker * wrapped when the service references */
this.invoker.invoke(new RpcInvocation(method, args)).recreate()
~~~
}
}
Copy the code2. During the invoke call, you need to use the Cluster, which is responsible for fault tolerance. The Cluster gets a list of all invokers of remote services that can be invoked from Directory before invoking them.
public class MockClusterInvoker<T> implements Invoker<T> {...public Result invoke(Invocation invocation) throws RpcException {
Result result = null;
// Get the mock configuration from the URL
String value = this.directory.getUrl()
.getMethodParameter(invocation.getMethodName(), "mock",Boolean.FALSE.toString()).trim();
if(value.length() ! =0 && !value.equalsIgnoreCase("false")) {
if (value.startsWith("force")) {
// Force the mock scene call
result = this.doMockInvoke(invocation, (RpcException)null);
} else {
// Fail to enter the mock scene
try {
result = this.invoker.invoke(invocation);
} catch (RpcException var5) {
// An exception occurs to enter the mock
result = this.doMockInvoke(invocation, var5); }}}else {
// No mock scene directly called
result = this.invoker.invoke(invocation);
}
return result;
}
Copy the code3. The invoker list obtained and the load balancer obtained through Dubbo SPI, Dubbo will invoke FailoverClusterInvoker doInvoke method.
public abstract class AbstractClusterInvoker<T> implements Invoker<T> {...public Result invoke(Invocation invocation) throws RpcException {
// Check whether the node is destroyed
this.checkWhetherDestroyed();
LoadBalance loadbalance = null;
/** * call Directory - call Rou * generate invoker object * get invoker list ***/
List<Invoker<T>> invokers = this.list(invocation);
if(invokers ! =null && !invokers.isEmpty()) {
// Gets the load balancing policy for the invoker of the first service provider, which defaults to Random
loadbalance = (LoadBalance)ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(((Invoker)invokers.get(0)).getUrl().getMethodParameter(RpcUtils.getMethodName(invocation), "loadbalance"."random"));
}
RpcUtils.attachInvocationIdIfAsync(this.getUrl(), invocation);
return this.doInvoke(invocation, invokers, loadbalance); }... }Copy the code4. Call the LoadBalance method for load balancing. Finally, select an Invoker that can be called and invoke the Invoker invoke method
public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
// invoker null check
this.checkInvokers(invokers, invocation);
// The number of retries. The default value is one
int len = this.getUrl().getMethodParameter(invocation.getMethodName(), "retries".2) + 1;
if (len <= 0) {
len = 1;
}
RpcException le = null;
List<Invoker<T>> invoked = new ArrayList(invokers.size());
Set<String> providers = new HashSet(len);
for(int i = 0; i < len; ++i) {
if (i > 0) {
// After a failure, it is necessary to reset whether the charm node is destroyed and whether the invokkers are empty
this.checkWhetherDestroyed();
copyinvokers = this.list(invocation);
this.checkInvokers(copyinvokers, invocation);
}
// Re-load balancing
Invoker<T> invoker = this.select(loadbalance, invocation, copyinvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers(invoked);
try {
// Make a remote call
Result result = invoker.invoke(invocation);
return var12;
} catch (RpcException var17) {
if (var17.isBiz()) {
throw var17;
}
le = var17;
} catch (Throwable var18) {
le = new RpcException(var18.getMessage(), var18);
} finally{ providers.add(invoker.getUrl().getAddress()); }} ···}}Copy the code5. The invoker object is actually InvokerWrapper, which wraps the ProtocolFilterWrapper class. This is wrapped in the Filter chain and the Listener chain. ConsumerContextFilter, FutureFilter, MonitorFilter, etc. The Listener chain is called after the wanfilter chain is called.
private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
Invoker<T> last = invoker;
// Obtain the automatically activated Filter based on the key and group
List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
if(! filters.isEmpty()) {// This is inverted traversal, because only inverted, the outermost Invoker can make the first filter
for (int i = filters.size() - 1; i >= 0; i--) {
final Filter filter = filters.get(i);
// The actual Invoker (service object ref) is placed at the end of the interceptor
final Invoker<T> next = last;
// Generate an exporter for each filter, strung together in turn
last = new Invoker<T>() {
...
@Override
public Result invoke(Invocation invocation) throws RpcException {
// $-- each call is passed to the next interceptor
returnfilter.invoke(next, invocation); }... }; }}return last;
}
Copy the codepublic class ConsumerContextFilter implements Filter {
public ConsumerContextFilter(a) {}public Result invoke(Invoker
invoker, Invocation invocation) throws RpcException {
RpcContext.getContext().setInvoker(invoker).setInvocation(invocation).setLocalAddress(NetUtils.getLocalHost(), 0).setRemoteAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort());
if (invocation instanceof RpcInvocation) {
((RpcInvocation)invocation).setInvoker(invoker);
}
RpcResult var4;
try {
RpcResult result = (RpcResult)invoker.invoke(invocation);
RpcContext.getServerContext().setAttachments(result.getAttachments());
var4 = result;
} finally {
RpcContext.getContext().clearAttachments();
}
returnvar4; }}Copy the code6. Invoke AbstractInvoker invoke (DubboInvoker doInvoker)
public abstract class AbstractInvoker<T> implements Invoker<T> {
public Result invoke(Invocation inv) throws RpcException {
if (this. Destroyed. The get ()) {...}else {
RpcInvocation invocation = (RpcInvocation)inv;
invocation.setInvoker(this); ...try {
return this.doInvoke(invocation);
} catch(InvocationTargetException var6) {...}catch(RpcException var7) {···}catch (Throwable var8) {
return newRpcResult(var8); }}}}Copy the code7. Processing logic of related parts of Dubbo protocol, including request construction and transmission. The underlying layer is the Exchange layer
public class DubboInvoker<T> extends AbstractInvoker<T> {
protected Result doInvoke(Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation)invocation;
String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment("path".this.getUrl().getPath());
inv.setAttachment("version".this.version);
ExchangeClient currentClient;
if (this.clients.length == 1) {
currentClient = this.clients[0];
} else {
currentClient = this.clients[this.index.getAndIncrement() % this.clients.length];
}
try {
boolean isAsync = RpcUtils.isAsync(this.getUrl(), invocation);
boolean isOneway = RpcUtils.isOneway(this.getUrl(), invocation);
int timeout = this.getUrl().getMethodParameter(methodName, "timeout".1000);
if (isOneway) {
boolean isSent = this.getUrl().getMethodParameter(methodName, "sent".false);
currentClient.send(inv, isSent);
RpcContext.getContext().setFuture((Future)null);
return new RpcResult();
} else if (isAsync) {
ResponseFuture future = currentClient.request(inv, timeout);
RpcContext.getContext().setFuture(new FutureAdapter(future));
return new RpcResult();
} else {
RpcContext.getContext().setFuture((Future)null);
return(Result)currentClient.request(inv, timeout).get(); }}catch(TimeoutException var9) {··}catch(RemotingException var10) {···}}Copy the code8.Dubbo’s Exchange layer and Transport layer are mainly related to network transmission and protocol
·· Ignore the protocol processing layer content
9. Enter the invoker method of the server ProtocolFilerWrapper class, and start the server logic processing. The filter and listener chains are wrapped, and the real method is called.
@Override
public Result invoke(Invoker
invoker, Invocation invocation) throws RpcException {
try {
Result result = invoker.invoke(invocation);
// GenericService invoker exceptions are not handled
if(result.hasException() && GenericService.class ! = invoker.getInterface()) {····}}}Copy the code10. Wrap class methods through InvokerWrapper.
11. In AbstractProxyInvoker. Invoke performs in doInvoker method. Here the default implementation of doInvoker generally has two JdkProxyFactory, JavassistProxyFactory, here is actually to get the implementation of the proxy class, and execute the business method of the proxy class, return the result.
/** * JdkRpcProxyFactory */
public class JdkProxyFactory extends AbstractProxyFactory {
@Override
@SuppressWarnings("unchecked")
public <T> T getProxy(Invoker
invoker, Class
[] interfaces)
{
return (T) Proxy.newProxyInstance(invoker.getInterface().getClassLoader(), interfaces, new InvokerInvocationHandler(invoker));
}
@Override
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName, Class
[] parameterTypes, Object[] arguments) throws Throwable {
Method method = proxy.getClass().getMethod(methodName, parameterTypes);
returnmethod.invoke(proxy, arguments); }}; }}Copy the codeTo summarize, the whole process of Dubbo Rpc service invocation, during the service reference process, Dubbo wraps the Invoker layer upon layer as a proxy object, which the user invokes as if it were a local service. Dubbo unlayers the Invoker layers of the proxy, each performing a specific function, and finally harmonizing to complete the invocation of the entire service.