How Does RPC Work? A Deep Dive into Java RPC Implementation with Netty and Zookeeper

What is RPC?

RPC (Remote Procedure Call) is a communication protocol that lets a program on one machine invoke a method on a program running on another machine as if it were a local call, abstracting away the underlying network details.

Basic RPC Call Flow

A typical RPC invocation involves the client generating a proxy for the service interface, discovering service addresses from a registry, sending a request over the network, the server decoding the request, invoking the target method via reflection, and returning the result to the client.

Service Interface Definition

public interface HelloService {
    String sayHello(String somebody);
}

Service Implementation

public class HelloServiceImpl implements HelloService {
    @Override
    public String sayHello(String somebody) {
        return "hello " + somebody + "!";
    }
}

Service Registration (Spring + Custom XML Namespace)

The service implementation is declared as a Spring bean and registered to a custom storm:service tag. The custom tag is defined in an XSD file placed under META-INF and linked to a StormServiceNamespaceHandler which creates a ProviderFactoryBean definition.

<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>

public class StormServiceNamespaceHandler extends NamespaceHandlerSupport {
    @Override
    public void init() {
        registerBeanDefinitionParser("service", new ProviderFactoryBeanDefinitionParser());
    }
}

public class ProviderFactoryBean implements FactoryBean, InitializingBean {
    private Class<?> serviceItf;
    private Object serviceObject;
    private String serverPort;
    private long timeout;
    private String appKey;
    private String groupName = "default";
    private int weight = 1;
    private int workerThreads = 10;
    // getters/setters omitted for brevity
    @Override
    public Object getObject() throws Exception {
        return serviceProxyObject; // proxy is created later
    }
    @Override
    public Class<?> getObjectType() {
        return serviceItf;
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        NettyServer.singleton().start(Integer.parseInt(serverPort));
        RegisterCenter.singleton().registerProvider(buildProviderServiceInfos());
    }
}

Netty Server Startup

public void start(final int port) {
    synchronized (NettyServer.class) {
        if (bossGroup != null || workerGroup != null) return;
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        ServerBootstrap b = new ServerBootstrap();
        b.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<SocketChannel>() {
             @Override
             protected void initChannel(SocketChannel ch) throws Exception {
                 ch.pipeline().addLast(new NettyDecoderHandler(StormRequest.class, serializeType));
                 ch.pipeline().addLast(new NettyEncoderHandler(serializeType));
                 ch.pipeline().addLast(new NettyServerInvokeHandler());
             }
         });
        channel = b.bind(port).sync().channel();
    }
}

Client Side (Consumer)

The consumer creates a JDK dynamic proxy for the service interface, obtains a list of provider nodes from Zookeeper, selects one according to a load‑balancing strategy, and sends a StormRequest via Netty. The response is wrapped in a StormResponse and returned to the caller.

public Object getProxy() {
    return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                                 new Class<?>[]{targetInterface}, this);
}

String serviceKey = targetInterface.getName();
List<ProviderService> providers = RegisterCenter.singleton()
        .getServiceMetaDataMap4Consume().get(serviceKey);
ClusterStrategy strategy = ClusterEngine.queryClusterStrategy(clusterStrategy);
ProviderService provider = strategy.select(providers);
InetSocketAddress address = new InetSocketAddress(provider.getServerIp(), provider.getServerPort());
Future<StormResponse> future = threadPool.submit(RevokerServiceCallable.of(address, request));
StormResponse response = future.get(request.getInvokeTimeout(), TimeUnit.MILLISECONDS);
return response != null ? response.getResult() : null;

Result Flow

After the server processes the request, it writes a StormResponse back through Netty. The client handler stores the response in a blocking queue so the original thread can retrieve the result synchronously.

@Override
protected void channelRead0(ChannelHandlerContext ctx, StormResponse response) throws Exception {
    RevokerResponseHolder.putResultValue(response);
}

Summary

The article walks through the entire RPC lifecycle: defining a service contract, registering the service with a Zookeeper‑based registry, starting a Netty server, handling serialization/deserialization, performing reflection‑based method invocation, and implementing a client that discovers services, applies load balancing, and synchronously obtains results. Full source code is available at https://github.com/fankongqiumu/storm.git .