Netty实现RPC案例

wen java案例 2026-08-05 266

本文目录导读：

Netty实现RPC案例

项目结构
Maven依赖 (pom.xml)
公共组件
服务端组件
客户端组件
API接口
测试客户端
运行说明
扩展功能

我来为您提供一个完整的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.java

Maven依赖 (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框架实现了：