本文目录导读：
我来为您提供一个完整的Netty实现RPC的案例,这个案例将展示如何使用Netty构建一个简单的RPC框架。
项目结构
netty-rpc/
├── pom.xml
├── src/main/java/
│ ├── com/rpc/
│ │ ├── common/
│ │ │ ├── RpcRequest.java
│ │ │ ├── RpcResponse.java
│ │ │ ├── RpcEncoder.java
│ │ │ └── RpcDecoder.java
│ │ ├── server/
│ │ │ ├── RpcServer.java
│ │ │ ├── ServiceProvider.java
│ │ │ └── RpcServerHandler.java
│ │ ├── client/
│ │ │ ├── RpcClient.java
│ │ │ ├── RpcProxy.java
│ │ │ └── RpcClientHandler.java
│ │ └── api/
│ │ ├── HelloService.java
│ │ └── HelloServiceImpl.javaMaven依赖 (pom.xml)
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.rpc</groupId>
<artifactId>netty-rpc</artifactId>
<version>1.0-SNAPSHOT</version>
<packaging>jar</packaging>
<properties>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.86.Final</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.13.4</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.7.36</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.2</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
公共组件
RpcRequest.java
package com.rpc.common;
import java.io.Serializable;
public class RpcRequest implements Serializable {
private String requestId;
private String className;
private String methodName;
private Class<?>[] parameterTypes;
private Object[] parameters;
private long timestamp;
// Getters and Setters
public String getRequestId() { return requestId; }
public void setRequestId(String requestId) { this.requestId = requestId; }
public String getClassName() { return className; }
public void setClassName(String className) { this.className = className; }
public String getMethodName() { return methodName; }
public void setMethodName(String methodName) { this.methodName = methodName; }
public Class<?>[] getParameterTypes() { return parameterTypes; }
public void setParameterTypes(Class<?>[] parameterTypes) { this.parameterTypes = parameterTypes; }
public Object[] getParameters() { return parameters; }
public void setParameters(Object[] parameters) { this.parameters = parameters; }
public long getTimestamp() { return timestamp; }
public void setTimestamp(long timestamp) { this.timestamp = timestamp; }
@Override
public String toString() {
return "RpcRequest{" +
"requestId='" + requestId + '\'' +
", className='" + className + '\'' +
", methodName='" + methodName + '\'' +
", timestamp=" + timestamp +
'}';
}
}
RpcResponse.java
package com.rpc.common;
import java.io.Serializable;
public class RpcResponse implements Serializable {
private String requestId;
private Object result;
private Throwable error;
private boolean success;
public boolean isSuccess() { return success; }
public void setSuccess(boolean success) { this.success = success; }
public String getRequestId() { return requestId; }
public void setRequestId(String requestId) { this.requestId = requestId; }
public Object getResult() { return result; }
public void setResult(Object result) { this.result = result; }
public Throwable getError() { return error; }
public void setError(Throwable error) { this.error = error; }
@Override
public String toString() {
return "RpcResponse{" +
"requestId='" + requestId + '\'' +
", result=" + result +
", success=" + success +
'}';
}
}
RpcEncoder.java
package com.rpc.common;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
public class RpcEncoder extends MessageToByteEncoder<Object> {
private final Class<?> genericClass;
private final ObjectMapper objectMapper = new ObjectMapper();
public RpcEncoder(Class<?> genericClass) {
this.genericClass = genericClass;
}
@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
if (genericClass.isInstance(msg)) {
byte[] data = objectMapper.writeValueAsBytes(msg);
out.writeInt(data.length);
out.writeBytes(data);
}
}
}
RpcDecoder.java
package com.rpc.common;
import com.fasterxml.jackson.databind.ObjectMapper;
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 final Class<?> genericClass;
private final ObjectMapper objectMapper = new ObjectMapper();
public RpcDecoder(Class<?> genericClass) {
this.genericClass = genericClass;
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if (in.readableBytes() < 4) {
return;
}
in.markReaderIndex();
int dataLength = in.readInt();
if (dataLength < 0) {
ctx.close();
return;
}
if (in.readableBytes() < dataLength) {
in.resetReaderIndex();
return;
}
byte[] data = new byte[dataLength];
in.readBytes(data);
Object obj = objectMapper.readValue(data, genericClass);
out.add(obj);
}
}
服务端组件
ServiceProvider.java
package com.rpc.server;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class ServiceProvider {
private static final Map<String, Object> serviceMap = new ConcurrentHashMap<>();
public static void register(Class<?> serviceInterface, Object serviceImpl) {
serviceMap.put(serviceInterface.getName(), serviceImpl);
System.out.println("Registered service: " + serviceInterface.getName());
}
public static Object getService(String className) {
Object service = serviceMap.get(className);
if (service == null) {
throw new RuntimeException("Service not found: " + className);
}
return service;
}
}
RpcServerHandler.java
package com.rpc.server;
import com.rpc.common.RpcRequest;
import com.rpc.common.RpcResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.lang.reflect.Method;
import java.util.UUID;
public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
RpcResponse response = new RpcResponse();
response.setRequestId(request.getRequestId());
try {
Object service = ServiceProvider.getService(request.getClassName());
Method method = service.getClass().getMethod(
request.getMethodName(),
request.getParameterTypes()
);
Object result = method.invoke(service, request.getParameters());
response.setResult(result);
response.setSuccess(true);
} catch (Throwable e) {
response.setError(e);
response.setSuccess(false);
e.printStackTrace();
}
ctx.writeAndFlush(response);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
RpcServer.java
package com.rpc.server;
import com.rpc.api.HelloService;
import com.rpc.api.HelloServiceImpl;
import com.rpc.common.RpcDecoder;
import com.rpc.common.RpcEncoder;
import com.rpc.common.RpcRequest;
import com.rpc.common.RpcResponse;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class RpcServer {
private final int port;
public RpcServer(int port) {
this.port = port;
}
public void start() throws Exception {
// 注册服务
ServiceProvider.register(HelloService.class, new HelloServiceImpl());
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new RpcDecoder(RpcRequest.class));
pipeline.addLast(new RpcEncoder(RpcResponse.class));
pipeline.addLast(new RpcServerHandler());
}
});
ChannelFuture future = bootstrap.bind(port).sync();
System.out.println("RPC Server started on port " + port);
future.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
new RpcServer(8080).start();
}
}
客户端组件
RpcClientHandler.java
package com.rpc.client;
import com.rpc.common.RpcRequest;
import com.rpc.common.RpcResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {
private final Map<String, CompletableFuture<RpcResponse>> pendingRequests = new ConcurrentHashMap<>();
private ChannelHandlerContext ctx;
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
this.ctx = ctx;
System.out.println("Connected to server");
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
CompletableFuture<RpcResponse> future = pendingRequests.remove(response.getRequestId());
if (future != null) {
future.complete(response);
}
}
public CompletableFuture<RpcResponse> sendRequest(RpcRequest request) {
CompletableFuture<RpcResponse> future = new CompletableFuture<>();
pendingRequests.put(request.getRequestId(), future);
ctx.writeAndFlush(request);
return future;
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
RpcClient.java
package com.rpc.client;
import com.rpc.common.RpcDecoder;
import com.rpc.common.RpcEncoder;
import com.rpc.common.RpcRequest;
import com.rpc.common.RpcResponse;
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;
import java.util.concurrent.CompletableFuture;
public class RpcClient {
private final String host;
private final int port;
private RpcClientHandler handler;
private EventLoopGroup group;
public RpcClient(String host, int port) {
this.host = host;
this.port = port;
}
public void connect() throws InterruptedException {
group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
handler = new RpcClientHandler();
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new RpcEncoder(RpcRequest.class));
pipeline.addLast(new RpcDecoder(RpcResponse.class));
pipeline.addLast(handler);
}
});
ChannelFuture future = bootstrap.connect(host, port).sync();
System.out.println("Connected to server at " + host + ":" + port);
}
public CompletableFuture<RpcResponse> call(RpcRequest request) {
return handler.sendRequest(request);
}
public void close() {
if (group != null) {
group.shutdownGracefully();
}
}
}
RpcProxy.java
package com.rpc.client;
import com.rpc.common.RpcRequest;
import com.rpc.common.RpcResponse;
import java.lang.reflect.Proxy;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
public class RpcProxy {
private final RpcClient client;
public RpcProxy(String host, int port) {
this.client = new RpcClient(host, port);
try {
client.connect();
} catch (InterruptedException e) {
throw new RuntimeException("Failed to connect to RPC server", e);
}
}
@SuppressWarnings("unchecked")
public <T> T create(Class<?> interfaceClass) {
return (T) Proxy.newProxyInstance(
interfaceClass.getClassLoader(),
new Class<?>[]{interfaceClass},
(proxy, method, args) -> {
RpcRequest request = new RpcRequest();
request.setRequestId(UUID.randomUUID().toString());
request.setClassName(interfaceClass.getName());
request.setMethodName(method.getName());
request.setParameterTypes(method.getParameterTypes());
request.setParameters(args);
request.setTimestamp(System.currentTimeMillis());
RpcResponse response = client.call(request)
.get(30, TimeUnit.SECONDS);
if (response.isSuccess()) {
return response.getResult();
} else {
throw new RuntimeException("RPC call failed", response.getError());
}
}
);
}
public void close() {
client.close();
}
}
API接口
HelloService.java
package com.rpc.api;
import java.io.Serializable;
public interface HelloService extends Serializable {
String sayHello(String name);
int add(int a, int b);
User getUser(String id);
}
User.java
package com.rpc.api;
import java.io.Serializable;
public class User implements Serializable {
private String id;
private String name;
private int age;
public User() {}
public User(String id, String name, int age) {
this.id = id;
this.name = name;
this.age = age;
}
// Getters and Setters
public String getId() { return id; }
public void setId(String id) { this.id = id; }
public String getName() { return name; }
public void setName(String name) { this.name = name; }
public int getAge() { return age; }
public void setAge(int age) { this.age = age; }
@Override
public String toString() {
return "User{" +
"id='" + id + '\'' +
", name='" + name + '\'' +
", age=" + age +
'}';
}
}
HelloServiceImpl.java
package com.rpc.api;
public class HelloServiceImpl implements HelloService {
@Override
public String sayHello(String name) {
return "Hello, " + name + "! Time: " + System.currentTimeMillis();
}
@Override
public int add(int a, int b) {
return a + b;
}
@Override
public User getUser(String id) {
return new User(id, "User-" + id, 25);
}
}
测试客户端
RpcClientTest.java
package com.rpc;
import com.rpc.api.HelloService;
import com.rpc.client.RpcProxy;
public class RpcClientTest {
public static void main(String[] args) {
RpcProxy proxy = new RpcProxy("localhost", 8080);
try {
// 创建代理对象
HelloService helloService = proxy.create(HelloService.class);
// 测试1: 简单字符串方法
System.out.println("\n=== Testing sayHello ===");
for (int i = 0; i < 5; i++) {
String result = helloService.sayHello("User" + i);
System.out.println("sayHello result: " + result);
}
// 测试2: 基本类型参数
System.out.println("\n=== Testing add ===");
int sum = helloService.add(10, 20);
System.out.println("10 + 20 = " + sum);
// 测试3: 返回对象
System.out.println("\n=== Testing getUser ===");
Object user = helloService.getUser("1001");
System.out.println("User info: " + user);
} finally {
proxy.close();
}
}
}
运行说明
启动服务端
mvn compile exec:java -Dexec.mainClass="com.rpc.server.RpcServer"
启动客户端
mvn compile exec:java -Dexec.mainClass="com.rpc.RpcClientTest"
扩展功能
添加负载均衡
public class LoadBalancer {
private static final Map<String, List<String>> serviceURLs = new ConcurrentHashMap<>();
private static final AtomicInteger counter = new AtomicInteger();
public static String getServiceURL(String serviceName) {
List<String> urls = serviceURLs.get(serviceName);
if (urls == null || urls.isEmpty()) {
throw new RuntimeException("No available service: " + serviceName);
}
int index = counter.getAndIncrement() % urls.size();
return urls.get(index);
}
}
添加服务注册发现
public class ServiceRegistry {
private static final Map<String, String> registry = new ConcurrentHashMap<>();
public static void register(String serviceName, String address) {
registry.put(serviceName, address);
}
public static String discover(String serviceName) {
return registry.get(serviceName);
}
}
这个RPC框架实现了：
- ✅ 完整的请求/响应编解码
- ✅ 动态代理调用
- ✅ 服务注册和管理
- ✅ 异步调用支持
- ✅ 异常处理
- ✅ 超时控制
可以根据实际需求扩展：
- 服务发现（ZooKeeper/Consul/ETCD）
- 负载均衡策略
- 熔断限流
- 链路追踪
- 序列化框架（Protobuf/Hessian/Kryo）
