grpc 编译结果图.png

辅助工具类

GrpcHelper

其中包含了Request、Response与JSONObject之间的转换等，主要是对应用的辅助。

package cn.com.yd.commons.grpc.util;

import cn.com.yd.commons.grpc.Request;
import cn.com.yd.commons.grpc.Response;
import com.alibaba.fastjson.JSONObject;
import com.google.protobuf.ByteString;
import io.protostuff.LinkedBuffer;
import io.protostuff.ProtostuffIOUtil;
import io.protostuff.runtime.RuntimeSchema;
import net.sf.cglib.proxy.Enhancer;
import net.sf.cglib.reflect.FastClass;
import net.sf.cglib.reflect.FastMethod;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class GrpcHelper {

    /**
     * 转换成GRPC方法需要的参数
     *
     * @param bean   Spring管理的bean实例名称
     * @param method 要执行的业务方法
     * @param args   业务方法的参数，可以为null
     * @return
     */
    public static Request toRequest(String bean, String method, Object[] args) {
        JSONObject json = new JSONObject();
        json.put("bean", bean);
        json.put("method", method);
        if (null != args) {
            json.put("args", args);
        }
        byte[] bytes = ProtobufUtils.serialize(json);
        return Request.newBuilder().setRequest(ByteString.copyFrom(bytes)).build();
    }

    /**
     * 将GRPC方法的请求参数转换成JSONObject
     *
     * @param request GRPC方法的请求参数
     * @return
     */
    public static JSONObject toJSONObject(Request request) {
        return ProtobufUtils.deserialize(request.getRequest().toByteArray(), JSONObject.class);
    }

    /**
     * 将响应结果转换成JSONObject
     * @param response 响应结果
     * @return
     */
    public static JSONObject toJSONObject(Response response) {
        return ProtobufUtils.deserialize(response.getReponse().toByteArray(), JSONObject.class);
    }


    public static JSONObject success() {
        JSONObject json = new JSONObject();
        json.put("state", true);
        json.put("msg", "成功");
        json.put("obj", null);
        return json;
    }

    /**
     * 执行具体的业务方法并返回执行结果
     *
     * @param bean   Spring管理的bean实例
     * @param method 要执行的业务方法
     * @param args   业务方法的参数，可以为null
     * @return
     */
    public static Object execute(Object bean, String method, Object[] args) {
        Object result = null;
        try {
            Enhancer enhancer = new Enhancer();
            enhancer.setSuperclass(bean.getClass());
            FastClass serviceFastClass = FastClass.create(bean.getClass());
            Class<?>[] argTypes = null;
            if (args != null) {
                argTypes = new Class[args.length];
                for (int i = 0; i < args.length; i++) {
                    Class<?> type = args[i].getClass();
                    argTypes[i] = type;
                }
            }
            FastMethod serviceFastMethod = serviceFastClass.getMethod(method, argTypes);
            result = serviceFastMethod.invoke(bean, args);
        } catch (Exception e) {
            result = e;
            e.printStackTrace();
        }
        return result;
    }

    /**
     * 将执行结果转换成Response
     * @param result 执行结果
     * @return
     */
    public static Response toResponse(Object result) {
        JSONObject json = null;
        if (result instanceof JSONObject) {
            json = (JSONObject) result;
        } else {
            json = new JSONObject();
            if (result instanceof Throwable) {
                json.put("state", false);
                json.put("msg", "失败");
                json.put("obj", ((Throwable) result).getMessage());
            } else {
                json.put("state", true);
                json.put("msg", "成功");
                json.put("obj", result);
            }
        }
        return Response.newBuilder().setReponse(ByteString.copyFrom(ProtobufUtils.serialize(json))).build();
    }
}

class ProtobufUtils {

    // 缓存 schema 对象的 map
    private static Map<Class<?>, RuntimeSchema<?>> cachedSchema = new ConcurrentHashMap<>();

    /**
     * 根据获取相应类型的schema方法
     */
    @SuppressWarnings({"unchecked", "unused"})
    private static <T> RuntimeSchema<T> getSchema(Class<T> clazz) {
        RuntimeSchema<T> schema = (RuntimeSchema<T>) cachedSchema.get(clazz);
        if (schema == null) {
            schema = RuntimeSchema.createFrom(clazz);
            cachedSchema.put(clazz, schema);
        }
        return schema;
    }

    /**
     * 序列化方法，将对象序列化为字节数组（对象 ---> 字节数组）
     */
    @SuppressWarnings("unchecked")
    public static <T> byte[] serialize(T obj) {
        Class<T> clazz = (Class<T>) obj.getClass();
        RuntimeSchema<T> schema = getSchema(clazz);
        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
        return ProtostuffIOUtil.toByteArray(obj, schema, buffer);
    }

    /**
     * 反序列化方法，将字节数组反序列化为对象（字节数组 ---> 对象）
     */
    public static <T> T deserialize(byte[] data, Class<T> clazz) {
        RuntimeSchema<T> schema = RuntimeSchema.createFrom(clazz);
        T message = schema.newMessage();
        ProtostuffIOUtil.mergeFrom(data, message, schema);
        return message;
    }
}

GrpcServer

package cn.com.yd.commons.grpc.util;

import io.grpc.BindableService;
import io.grpc.Server;
import io.grpc.ServerBuilder;

import java.io.IOException;

public class GrpcServer {
    private final Server server;//服务器

    public GrpcServer(int port, BindableService... services) throws IOException {
        //初始化Server参数
        ServerBuilder builder = ServerBuilder.forPort(port);
        for(BindableService bs:services){
            builder.addService(bs);
        }
        server = builder.build();
    }

    /**
     * 启动服务
     */
    public void start() throws IOException {
        server.start();
        System.out.println("Server started, listening on " + server.getPort());
        //程序退出时关闭资源
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.err.println("*** shutting down gRPC server since JVM is shutting down");
            GrpcServer.this.stop();
            System.err.println("*** server shut down");
        }));
    }

    /**
     * 关闭服务
     */
    public void stop() {
        if (server != null) {
            server.shutdown();
        }
    }

    /**
     * 使得server一直处于运行状态
     */
    public void blockUntilShutdown() throws InterruptedException {
        if (server != null) {
            server.awaitTermination();
        }
    }
}

GrpcClient

import cn.com.yd.commons.grpc.CommonServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import java.util.concurrent.TimeUnit;
public class GrpcClient {
    private static ManagedChannel channel;//grpc信道,需要指定端口和地址
    private static CommonServiceGrpc.CommonServiceBlockingStub blockingStub;//阻塞/同步存根
    private static CommonServiceGrpc.CommonServiceStub asyncStub;//非阻塞,异步存根
    public GrpcClient(String server, int port) {
        System.out.println("server:"+server+",port:"+port);
        //创建信道
        channel = ManagedChannelBuilder.forAddress(server, port)
                .usePlaintext()
                .build();
        //创建存根
        blockingStub = CommonServiceGrpc.newBlockingStub(channel);
        asyncStub = CommonServiceGrpc.newStub(channel);
    }

    /**
     * 关闭方法
     */
    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }

    public CommonServiceGrpc.CommonServiceBlockingStub getBlockingStub() {
        return blockingStub;
    }

    public CommonServiceGrpc.CommonServiceStub getAsyncStub() {
        return asyncStub;
    }
}

工程源码

暂不公开。