位置：首页-资讯-后端开发

基于Springboot+Netty实现rpc的方法附demo

2024-04-02 19:55

短信预约 -IT技能 免费直播动态提醒

今天翻看了一下Netty相关的知识点，正好练练手，简单捣鼓了这个demo;这里简单梳理一下；

前提知识点：

Springboot、 Netty、动态代理(反射)、反射

项目整体结构如下：

1.在父项目中引入相关依赖；

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.3.2.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.48.Final</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.58</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>2.0.0-alpha1</version>
        </dependency>

2.服务提供模块整体结构如下：

这里重点关注一下 RequestModel 和 ResponseModel 两个消息体类，

@Data
@AllArgsConstructor
public class RequestModel {
 
    private String requestId;
    private String serviceName;
    private String methodName;
    private Class[] paramTypes;
    private Object[] paramValues;
 
}

@Data
@AllArgsConstructor
public class ResponseModel {
    private String responseId;
    private String serviceName;
    private String methodName;
    private String code;
    private String data;
}

用于服务端和客户端的数据传输；再者就是关注 ServerChannelInboundHandler 中的 channelRead0() 报文解码处理；

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        StringBuilder sb = null;
        RequestModel result = null;
        try {
            // 报文解析处理
            sb = new StringBuilder();
            result = JSON.parseObject(msg, RequestModel.class);
 
            requestId = result.getRequestId();
            String serviceName = result.getServiceName();
            String methodName = result.getMethodName();
            Class[] paramType = result.getParamTypes();
            Object[] paramValue = result.getParamValues();
            System.out.println(serviceName + "  " + methodName);
            String substring = serviceName.substring(serviceName.lastIndexOf(".") + 1);
            String s = substring.substring(0, 1).toLowerCase() + substring.substring(1);
            Object serviceObject = applicationContext.getBean(s);
            Method method = Class.forName(serviceName).getMethod(methodName, paramType);
            Object returnValue = method.invoke(serviceObject, paramValue);
            ResponseModel responseModel = new ResponseModel(requestId,serviceName,methodName,"200",JSON.toJSONString(returnValue));
            sb.append(JSON.toJSONString(responseModel));
            sb.append("\n");
            System.out.println(sb.toString());
            ctx.writeAndFlush(sb);
        } catch (Exception e) {
            ResponseModel responseModel = new ResponseModel(requestId,"","","500",e.getMessage());
            String errorCode = JSON.toJSONString(responseModel)+"\n";
            log.error(errorCode);
            ctx.writeAndFlush(errorCode);
            log.error("报文解析失败: " + e.getMessage());
        }
    }

客户端的模块代码如下；

这里重点关注的是 ClientHandler 类中 channelRead0() 方法的处理

 @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        System.out.println("收到服务端消息: " + msg);
 
        ResponseModel responseModel = JSON.parseObject(msg,ResponseModel.class);
        String responseId = responseModel.getResponseId();
        Promise promise = LocalPromise.promiseMap.remove(responseId);
        if(promise != null){
            String code = responseModel.getCode();
            if(code.equals("200")){
                promise.setSuccess(responseModel.getData());
            }else{
                promise.setFailure(new RuntimeException(responseModel.getData()));
            }
        }
    }

和 AppStart 类中获取获取服务的处理；

private <T> T getProxyService(Class<T> serviceClass) {
        Object service = Proxy.newProxyInstance(serviceClass.getClassLoader(), new Class[]{serviceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                Channel channel = NettyClient.getChannel(host, port);
                RequestModel requestModel = new RequestModel("100001", method.getDeclaringClass().getName(), method.getName(), method.getParameterTypes(), args);
                channel.writeAndFlush(JSON.toJSONString(requestModel) + "\n");
                Promise promise = new DefaultPromise(channel.eventLoop());
                LocalPromise.promiseMap.put(requestModel.getRequestId(), promise);
 
                System.out.println(LocalPromise.promiseMap+">>>>>>>>>>>>");
                promise.await();
                if (promise.isSuccess()) {
                    Class<?> returnType = method.getReturnType();
                    return JSON.toJavaObject(JSON.parseObject(promise.getNow()+""),returnType);
                } else {
                    System.out.println(promise.cause());
                    return promise.cause();
                }
            }
        });
        return (T) service;
    }

测试结果：