5.3 编辑客户端和服务段代码

5.3.1 客户端代码

package io.grpc.transport;

import com.ahhx.jhpt.server.base.receiver.SynFilesReceiver;
import com.ahhx.jhpt.server.base.receiver.SynOriginReceiver;
import com.ahhx.jhpt.server.base.rpc.SjtbClient;
import com.ahhx.jhpt.server.base.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author zhanglei
 * @email ah.zhanglei4@aisino.com
 * @date 2019/7/8
 */
public class TransportClient {
    private static final Logger logger = Logger.getLogger(TransportClient.class.getName());

    private final ManagedChannel channel;
    private final MessageGuideGrpc.MessageGuideBlockingStub blockingStub;
    private final MessageGuideGrpc.MessageGuideStub asyncStub;

    private static TransportClient instance = null;

    public static String grpcServerIp;
    public static int grpcServerPort;


    /** Issues several different requests and then exits. */
//    @Test
//    public void test() throws InterruptedException {
//        List<Message> messages = new ArrayList<Message>();
//        for (int i = 0; i < 1; i++) {
//            Message m = Message.newBuilder().setData("111"+i).setType("type"+i).build();
//            messages.add(m);
//        }
//
//        TransportClient client = new TransportClient();
//        try {
//            client.recordRoute(messages, null, null, null, 0, null);
//        } finally {
//            client.shutdown();
//        }
//    }


    public void send(String methodName, String body, SynFilesReceiver synFilesReceiver, SynOriginReceiver synOriginReceiver, Channel mqchannel, long tag, org.slf4j.Logger logger) throws Exception {
        List<Message> messages = new ArrayList<Message>();
        ////String body2 = EncodeUtils.unCompressAndDecode(body);
        Message m = Message.newBuilder().setData(body).setType(methodName).build();
        messages.add(m);
        try {
            instance.recordRoute(messages,synFilesReceiver,synOriginReceiver,mqchannel,tag,logger);
        } finally {
            ////instance.shutdown();
            ////instance = null;
        }
    }

    public static TransportClient newInstance() {
        if (instance == null) {
            synchronized (SjtbClient.class){
                if (instance == null) {
                    instance = new TransportClient();
                }
            }
        }
        return instance;
    }

    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }

    /**
     * Async client-streaming example. Sends {@code numPoints} randomly chosen points from {@code
     * features} with a variable delay in between. Prints the statistics when they are sent from the
     * server.
     */
    public void recordRoute(List<Message> messages, SynFilesReceiver synFilesReceiver, SynOriginReceiver synOriginReceiver, Channel mqchannel, long tag, org.slf4j.Logger logger) throws InterruptedException {
        info("*** RecordRoute");

        StreamObserver<BackMessage> responseObserver = new StreamObserver<BackMessage>() {

            @Override
            public void onNext(BackMessage backMessage) {
                TransportClient.logger.info("backMessage:"+backMessage);
                if (null != synFilesReceiver){
                    synFilesReceiver.replyToSender(backMessage.getData());
                    TransportClient.logger.info("处理数据包到hdfs接口服务完成...");
                }
                if (null != synOriginReceiver){
                    synOriginReceiver.replyToSender(backMessage.getData());
                    TransportClient.logger.info("处理写表操作调用WriteTable接口服务完成...");
                }
            }

            @Override
            public void onError(Throwable throwable) {
                RabbitMqUtils.rejectAndBackMQ(mqchannel, tag, logger);
                warning("RecordRoute Failed: {0}", Status.fromThrowable(throwable));
            }

            @Override
            public void onCompleted() {
                info("Finished RecordRoute");
                if (null != synFilesReceiver){
                    RabbitMqUtils.askMessage(mqchannel, tag, logger);
                }
            }
        };

        StreamObserver<Message> requestObserver = asyncStub.recordRoute(responseObserver);
        try {
            for (int i = 0; i < messages.size(); i++) {
                requestObserver.onNext(messages.get(i));
            }
        } catch (RuntimeException e) {
            //// Cancel RPC
            requestObserver.onError(e);
            throw e;
        }
        //// Mark the end of requests
        requestObserver.onCompleted();
    }

    /** Construct client for accessing RouteGuide server using the existing channel. */
    public TransportClient() {
        channel =  ManagedChannelBuilder.forAddress(grpcServerIp, grpcServerPort).usePlaintext().build();
        blockingStub = MessageGuideGrpc.newBlockingStub(channel);
        asyncStub = MessageGuideGrpc.newStub(channel);
    }

    private void info(String msg, Object... params) {
        logger.log(Level.INFO, msg, params);
    }

    private void warning(String msg, Object... params) {
        logger.log(Level.WARNING, msg, params);
    }

}

调用asyncStub.recordRoute可以获得一个requestObserver对象，requestObserver的onNext就可以将数据发送出去。
此外，构建一个StreamObserver，这个类中的回掉方法会在服务段返回数据回来的时候被回掉。

5.3.2 服务端代码

package io.grpc.transport;

import com.ahhx.jhpt.server.base.service.SynDataService;
import com.ahhx.jhpt.server.base.utils.SpringContextUtil;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import org.junit.Test;

import java.io.IOException;
import java.lang.reflect.Method;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zhanglei
 * @email ah.zhanglei4@aisino.com
 * @date 2019/7/8
 */
public class TransportServer {

    private SynDataService synDataService = SpringContextUtil.getBean(SynDataService.class);
    private static final Logger logger = LoggerFactory.getLogger(TransportServer.class.getName());
    private Server server;

    private int grpcServerPort;

    public TransportServer(int grpcServerPort) {
        this.grpcServerPort = grpcServerPort;
        server = ServerBuilder.forPort(grpcServerPort).addService(new TransportService())
                .build();
    }

    @Test
    public void test() throws Exception {
        TransportServer server = new TransportServer(20002);
        server.start();
        server.blockUntilShutdown();
    }

    public void blockUntilShutdown() throws InterruptedException {
        if (server != null) {
            server.awaitTermination();
        }
    }

    /** Start serving requests. */
    public void start() throws IOException {
        server.start();
        logger.info("Server started, listening on " + grpcServerPort);
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                // Use stderr here since the logger may have been reset by its JVM shutdown hook.
                System.err.println("*** shutting down gRPC server since JVM is shutting down");
                TransportServer.this.stop();
                System.err.println("*** server shut down");
            }
        });
    }


    private class TransportService extends MessageGuideGrpc.MessageGuideImplBase {
        final long startTime = System.nanoTime();
        @Override
        public StreamObserver<Message> recordRoute(StreamObserver<BackMessage> responseObserver) {

            return new StreamObserver<Message>() {
                @Override
                public void onNext(Message message) {
                    logger.info("message:"+message);
                    logger.info("开始处理业务逻辑.");
                    try {
                        Method method = SynDataService.class.getMethod(message.getType(), String.class);
                        String backMessageData = (String) method.invoke(synDataService, message.getData());
                        responseObserver.onNext(BackMessage.newBuilder().setData(backMessageData)
                                .build());
                    } catch (Exception e) {
                        logger.info("调用方法异常：" + e.getMessage());
                    }finally {
                        logger.info("处理业务逻辑完成.");
                    }
                }

                @Override
                public void onError(Throwable throwable) {
                    responseObserver.onError(throwable);
                }

                @Override
                public void onCompleted() {
                    responseObserver.onCompleted();
                }
            };
        }
    }

    /** Stop serving requests and shutdown resources. */
    public void stop() {
        if (server != null) {
            server.shutdown();
        }
    }

}

构建server对象的时候addservice，这个service类中的回掉方法，会在接收到客户端请求时被回掉。

6. 结束

在idea中，通过点击下面，可以生成java代码

实测，在linux中执行maven打包的时候会自动生成java代码，生成jar包。

更过关于GRPC的内容，推荐链接：
http://doc.oschina.net/grpc?t=56831
https://github.com/grpc/grpc
https://github.com/grpc/grpc-java

版权声明：本文来源简书，感谢博主原创文章，遵循 CC 4.0 by-sa 版权协议，转载请附上原文出处链接和本声明。
原文链接：https://www.jianshu.com/p/292c16558b82
站方申明：本站部分内容来自社区用户分享，若涉及侵权，请联系站方删除。

• 发表于 2020-02-02 14:52:50
• 阅读 ( 1473 )
• 分类：