java rpc netty框架代码
时间: 2024-03-15 14:35:07 浏览: 21
以下是一个简单的 Java RPC Netty 框架代码实现:
```java
// 定义 RPC 请求实体类
public class RpcRequest implements Serializable {
private String className;
private String methodName;
private Object[] parameters;
// 省略 getter 和 setter 方法
}
// 定义 RPC 响应实体类
public class RpcResponse implements Serializable {
private Object result;
private String error;
// 省略 getter 和 setter 方法
}
// 定义 RPC 服务接口
public interface RpcService {
// 定义服务方法
public int add(int a, int b);
}
// 实现 RPC 服务接口
public class RpcServiceImpl implements RpcService {
@Override
public int add(int a, int b) {
return a + b;
}
}
// 定义 RPC 服务器端
public class RpcServer {
private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);
private String host;
private int port;
public RpcServer(String host, int port) {
this.host = host;
this.port = port;
}
public void start() throws InterruptedException {
// 创建线程池
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
// 配置服务器启动类
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
// 添加解码器和编码器
ch.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
ch.pipeline().addLast(new ObjectEncoder());
// 添加业务处理类
ch.pipeline().addLast(new RpcServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
// 启动服务器
ChannelFuture f = b.bind(host, port).sync();
LOGGER.info("Server started on {}:{}", host, port);
// 等待直到服务器 socket 关闭
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
// 定义业务处理类
private class RpcServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
RpcRequest request = (RpcRequest) msg;
LOGGER.info("Received request: {}", request);
// 调用服务
RpcResponse response = new RpcResponse();
try {
Class<?> clazz = Class.forName(request.getClassName());
Method method = clazz.getMethod(request.getMethodName(), request.getParameters().getClass());
Object result = method.invoke(clazz.newInstance(), request.getParameters());
LOGGER.info("Result: {}", result);
response.setResult(result);
} catch (Exception e) {
LOGGER.error("Error occurred while invoking service: {}", e.getMessage());
response.setError(e.getMessage());
}
// 返回响应
ctx.writeAndFlush(response);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
LOGGER.error("Error occurred in server: {}", cause.getMessage());
ctx.close();
}
}
}
// 定义 RPC 客户端
public class RpcClient {
private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);
private String host;
private int port;
public RpcClient(String host, int port) {
this.host = host;
this.port = port;
}
public Object invoke(String className, String methodName, Object... parameters) {
// 创建连接
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
// 添加解码器和编码器
ch.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
ch.pipeline().addLast(new ObjectEncoder());
// 添加业务处理类
ch.pipeline().addLast(new RpcClientHandler());
}
})
.option(ChannelOption.SO_KEEPALIVE, true);
// 连接服务器
ChannelFuture f = b.connect(host, port).sync();
LOGGER.info("Connected to server {}:{}", host, port);
// 发送请求
RpcRequest request = new RpcRequest();
request.setClassName(className);
request.setMethodName(methodName);
request.setParameters(parameters);
f.channel().writeAndFlush(request);
// 等待服务器响应
RpcResponse response = RpcClientHandler.getResponse();
if (response.getError() != null) {
LOGGER.error("Error occurred while invoking service: {}", response.getError());
} else {
LOGGER.info("Result: {}", response.getResult());
return response.getResult();
}
// 关闭连接
f.channel().closeFuture().sync();
} catch (Exception e) {
LOGGER.error("Error occurred in client: {}", e.getMessage());
} finally {
group.shutdownGracefully();
}
return null;
}
// 定义业务处理类
private static class RpcClientHandler extends ChannelInboundHandlerAdapter {
private static RpcResponse response;
public static RpcResponse getResponse() {
return response;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
response = (RpcResponse) msg;
LOGGER.info("Received response: {}", response);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
LOGGER.error("Error occurred in client: {}", cause.getMessage());
ctx.close();
}
}
}
// 测试代码
public class RpcTest {
public static void main(String[] args) throws InterruptedException {
// 启动服务器
RpcServer server = new RpcServer("localhost", 8080);
server.start();
// 等待服务器启动
Thread.sleep(1000);
// 创建客户端并调用服务
RpcClient client = new RpcClient("localhost", 8080);
RpcService service = (RpcService) client.invoke("RpcServiceImpl", "add", 1, 2);
System.out.println(service.add(1, 2));
}
}
```
以上代码实现了一个简单的 Java RPC Netty 框架,包括 RPC 请求和响应实体类、RPC 服务接口和实现类、RPC 服务器端和客户端。其中,RPC 服务器端和客户端都使用了 Netty 框架,通过序列化和反序列化实现了数据的传输和处理。在测试代码中,我们启动了一个 RPC 服务器,并通过客户端调用了 RPC 服务。