devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

grpc java io通信模型_io.grpc.managedchannel

作者：我家小花儿 | 2024-04-08 02:46:34

赞

踩

io.grpc.managedchannel
1 gRPC 服务调用
gRPC 的通信协议基于标准的 HTTP/2 设计，主要提供了两种 RPC 调用方式：
1   普通 RPC 调用方式，即请求 - 响应模式。

2  基于 HTTP/2.0 的 streaming 调用方式。



1.1 普通 RPC 调用
普通的 RPC 调用提供了三种实现方式：
1  同步阻塞式服务调用，通常实现类是 xxxBlockingStub（基于 proto 定义生成）。
2  异步非阻塞调用，基于 Future-Listener 机制，通常实现类是 xxxFutureStub。

3  异步非阻塞调用，基于 Reactive(Async) 的响应式编程模式，通常实现类是 xxxStub。


普通 rpc demo 如下：

  1. import java.util.concurrent.TimeUnit;
  2.  
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5.  
  6. import com.ylifegroup.protobuf.PhoneServiceGrpc;
  7. import com.ylifegroup.protobuf.Phonebook.AddPhoneToUserRequest;
  8. import com.ylifegroup.protobuf.Phonebook.AddPhoneToUserResponse;
  9. import com.ylifegroup.protobuf.Phonebook.PhoneType;
  10.  
  11. import io.grpc.ManagedChannel;
  12. import io.grpc.ManagedChannelBuilder;
  13. import io.grpc.StatusRuntimeException;
  14.  
  15. /**
  16.  * @describe GRpcClient Block demo
  17.  * @author zhikai.chen
  18.  * @date 2018年5月7日 下午4:00:58
  19.  */
  20. public class GRpcClientBlock {
  21. 	
  22. 	private static final Logger logger = LoggerFactory.getLogger(GRpcClientBlock.class);
  23.  
  24.     private final ManagedChannel channel;
  25.  
  26.     private final PhoneServiceGrpc.PhoneServiceBlockingStub blockingStub;
  27.  
  28.     /** Construct client connecting to gRPC server at {@code host:port}. */
  29.     public GRpcClientBlock(String host, int port) {
  30.         ManagedChannelBuilder<?> channelBuilder = ManagedChannelBuilder.forAddress(host, port).usePlaintext();
  31.         channel = channelBuilder.build();
  32.         blockingStub = PhoneServiceGrpc.newBlockingStub(channel);
  33.     }
  34.  
  35.     public void shutdown() throws InterruptedException {
  36.         channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
  37.     }
  38.  
  39.     /** add phone to user. */
  40.     public void addPhoneToUser(int uid, PhoneType phoneType, String phoneNubmer) {
  41.         logger.info("Will try to add phone to user " + uid);
  42.         AddPhoneToUserRequest request = AddPhoneToUserRequest.newBuilder().setUid(uid).setPhoneType(phoneType)
  43.                 .setPhoneNumber(phoneNubmer).build();
  44.         AddPhoneToUserResponse response;
  45.         try {
  46.             response = blockingStub.addPhoneToUser(request);
  47.         } catch (StatusRuntimeException e) {
  48.             logger.warn("RPC failed: {0} --> "+e.getLocalizedMessage(), e.getStatus());
  49.             return;
  50.         }
  51.         logger.info("Result: " + response.getResult());
  52.     }
  53.  
  54.     public static void main(String[] args) throws Exception {
  55.         GRpcClientBlock client = new GRpcClientBlock("localhost", 50051);
  56.         try {
  57.             client.addPhoneToUser(1, PhoneType.WORK, "13888888888");
  58.         } finally {
  59.             client.shutdown();
  60.         }
  61.     }
  62.  
  63. }


Future 模型demo如下：

  1. import java.util.concurrent.Executors;
  2. import java.util.concurrent.TimeUnit;
  3.  
  4. import javax.annotation.Nullable;
  5.  
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  8.  
  9. import com.google.common.util.concurrent.FutureCallback;
  10. import com.google.common.util.concurrent.Futures;
  11. import com.ylifegroup.protobuf.PhoneServiceGrpc;
  12. import com.ylifegroup.protobuf.Phonebook.AddPhoneToUserRequest;
  13. import com.ylifegroup.protobuf.Phonebook.AddPhoneToUserResponse;
  14. import com.ylifegroup.protobuf.Phonebook.PhoneType;
  15.  
  16. import io.grpc.ManagedChannel;
  17. import io.grpc.ManagedChannelBuilder;
  18.  
  19. /**
  20.  * @describe GRpcClient Future demo
  21.  * @author zhikai.chen
  22.  * @date 2018年5月7日 下午4:00:58
  23.  */
  24. public class GRpcClientFuture {
  25. 	
  26. 	private static final Logger logger = LoggerFactory.getLogger(GRpcClientFuture.class);
  27.  
  28.     private final ManagedChannel channel;
  29.  
  30.     private final PhoneServiceGrpc.PhoneServiceFutureStub futureStub;
  31.  
  32.     /** Construct client connecting to gRPC server at {@code host:port}. */
  33.     public GRpcClientFuture(String host, int port) {
  34.         ManagedChannelBuilder<?> channelBuilder = ManagedChannelBuilder.forAddress(host, port).usePlaintext();
  35.         channel = channelBuilder.build();
  36.         futureStub = PhoneServiceGrpc.newFutureStub(channel);
  37.     }
  38.  
  39.     public void shutdown() throws InterruptedException {
  40.         channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
  41.     }
  42.  
  43.     /** add phone to user. */
  44.     public void addPhoneToUserFuture1(int uid, PhoneType phoneType, String phoneNubmer) {
  45.         logger.info("Will try to add phone to user " + uid);
  46.         AddPhoneToUserRequest request = AddPhoneToUserRequest.newBuilder().setUid(uid).setPhoneType(phoneType)
  47.         		.setPhoneNumber(phoneNubmer).build();
  48.  
  49.         try {
  50.         	com.google.common.util.concurrent.ListenableFuture<AddPhoneToUserResponse>
  51.         	listenableFuture = futureStub.addPhoneToUser(request);
  52.         	Futures.addCallback(listenableFuture, new FutureCallback<AddPhoneToUserResponse>() {
  53.         		@Override
  54.         		public void onSuccess(@Nullable AddPhoneToUserResponse result) {
  55.         			logger.info("result: " + result.getResult());
  56.         		}
  57.         		@Override
  58.         		public void onFailure(Throwable t) {
  59.         			logger.warn(t.getMessage());
  60.         		}
  61.         	});
  62.         } catch (Exception e) {
  63.         	logger.warn("RPC failed: {0}", e);
  64.         	return;
  65.         }
  66.     }
  67.     
  68.     /** add phone to user. */
  69.     public void addPhoneToUserFuture2(int uid, PhoneType phoneType, String phoneNubmer) {
  70.         logger.info("Will try to add phone to user " + uid);
  71.         AddPhoneToUserRequest request = AddPhoneToUserRequest.newBuilder().setUid(uid).setPhoneType(phoneType)
  72.         		.setPhoneNumber(phoneNubmer).build();
  73.  
  74.         try {
  75.         	com.google.common.util.concurrent.ListenableFuture<AddPhoneToUserResponse>
  76.         	listenableFuture = futureStub.addPhoneToUser(request);
  77.         	listenableFuture.addListener(()->
  78.         	{
  79.         		try {
  80.         			AddPhoneToUserResponse response = listenableFuture.get();
  81.         			logger.info("result: " + response.getResult());
  82.         		}
  83.         		catch(Exception e)
  84.         		{
  85.         			e.printStackTrace();
  86.         		}
  87.         	}, Executors.newFixedThreadPool(1));
  88.         } catch (Exception e) {
  89.         	logger.warn("RPC failed: {0}", e);
  90.         	return;
  91.         }
  92.  
  93.     }
  94.  
  95.     public static void main(String[] args) throws Exception {
  96.         GRpcClientFuture client = new GRpcClientFuture("localhost", 50051);
  97.         try {
  98.             client.addPhoneToUserFuture1(1, PhoneType.WORK, "13888888888");
  99.             client.addPhoneToUserFuture2(2, PhoneType.WORK, "13888888888");
  100.             TimeUnit.SECONDS.sleep(3);
  101.             //TODO 这个模式需要自己手动退出
  102.             Runtime.getRuntime().exit(0);
  103.         } finally {
  104.             client.shutdown();
  105.         }
  106.     }
  107.  
  108. }

Reactive(Async)模型demo如下：

  1. import java.util.concurrent.TimeUnit;
  2.  
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5.  
  6. import com.ylifegroup.protobuf.PhoneServiceGrpc;
  7. import com.ylifegroup.protobuf.Phonebook.AddPhoneToUserRequest;
  8. import com.ylifegroup.protobuf.Phonebook.AddPhoneToUserResponse;
  9. import com.ylifegroup.protobuf.Phonebook.PhoneType;
  10.  
  11. import io.grpc.ManagedChannel;
  12. import io.grpc.ManagedChannelBuilder;
  13.  
  14. /**
  15.  * @describe GRpcClient Async demo
  16.  * @author zhikai.chen
  17.  * @date 2018年5月7日 下午4:00:58
  18.  */
  19. public class GRpcClientAsync {
  20. 	
  21.     private static final Logger logger = LoggerFactory.getLogger(GRpcClientAsync.class);
  22.  
  23.     private final ManagedChannel channel;
  24.  
  25.     private final PhoneServiceGrpc.PhoneServiceStub stub;
  26.  
  27.     /** Construct client connecting to gRPC server at {@code host:port}. */
  28.     public GRpcClientAsync(String host, int port) {
  29.         ManagedChannelBuilder<?> channelBuilder = ManagedChannelBuilder.forAddress(host, port).usePlaintext();
  30.         channel = channelBuilder.build();
  31.         stub = PhoneServiceGrpc.newStub(channel);
  32.     }
  33.  
  34.     public void shutdown() throws InterruptedException {
  35.         channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
  36.     }
  37.  
  38.     /** add phone to user. */
  39.     public void addPhoneToUserAsync(int uid, PhoneType phoneType, String phoneNubmer) {
  40.         logger.info("Will try to add phone to user " + uid);
  41.         AddPhoneToUserRequest request = AddPhoneToUserRequest.newBuilder().setUid(uid).setPhoneType(phoneType)
  42.                 .setPhoneNumber(phoneNubmer).build();
  43.         
  44.         io.grpc.stub.StreamObserver<AddPhoneToUserResponse> responseObserver =
  45.         		new io.grpc.stub.StreamObserver<AddPhoneToUserResponse>()
  46.         {
  47.         	public  void onNext(AddPhoneToUserResponse response)
  48.         	{
  49.         		logger.info("Result: " + response.getResult());
  50.         	}
  51.         	public void onError(Throwable t){
  52.         		logger.warn(t.getMessage());
  53.         	}
  54.         	public void onCompleted(){}
  55.         };
  56.         stub.addPhoneToUser(request,responseObserver);
  57.     }
  58.  
  59.     public static void main(String[] args) throws Exception {
  60.         GRpcClientAsync client = new GRpcClientAsync("localhost", 50051);
  61.         try {
  62.             client.addPhoneToUserAsync(1, PhoneType.WORK, "13888888888");
  63.             TimeUnit.SECONDS.sleep(5);
  64.         } finally {
  65.             client.shutdown();
  66.         }
  67.     }
  68.  
  69. }


1.2 Streaming 模式服务调用
基于 HTTP/2.0,gRPC 提供了三种 streaming 模式：
1  服务端 streaming
2  客户端 streaming

3  服务端和客户端双向 streaming

服务端streaming demo如下：

  1. import static java.lang.Math.atan2;
  2. import static java.lang.Math.cos;
  3. import static java.lang.Math.max;
  4. import static java.lang.Math.min;
  5. import static java.lang.Math.sin;
  6. import static java.lang.Math.sqrt;
  7. import static java.lang.Math.toRadians;
  8. import static java.util.concurrent.TimeUnit.NANOSECONDS;
  9.  
  10. import io.grpc.Server;
  11. import io.grpc.ServerBuilder;
  12. import io.grpc.examples.routeguide.Feature;
  13. import io.grpc.examples.routeguide.Point;
  14. import io.grpc.examples.routeguide.Rectangle;
  15. import io.grpc.examples.routeguide.RouteGuideGrpc;
  16. import io.grpc.examples.routeguide.RouteNote;
  17. import io.grpc.examples.routeguide.RouteSummary;
  18. import io.grpc.stub.StreamObserver;
  19. import java.io.IOException;
  20. import java.net.URL;
  21. import java.util.ArrayList;
  22. import java.util.Collection;
  23. import java.util.Collections;
  24. import java.util.List;
  25. import java.util.concurrent.ConcurrentHashMap;
  26. import java.util.concurrent.ConcurrentMap;
  27. import java.util.logging.Level;
  28. import java.util.logging.Logger;
  29.  
  30. /**
  31.  * A sample gRPC server that serve the RouteGuide (see route_guide.proto) service.
  32.  */
  33. public class RouteGuideServer {
  34.   private static final Logger logger = Logger.getLogger(RouteGuideServer.class.getName());
  35.  
  36.   private final int port;
  37.   private final Server server;
  38.  
  39.   public RouteGuideServer(int port) throws IOException {
  40.     this(port, RouteGuideUtil.getDefaultFeaturesFile());
  41.   }
  42.  
  43.   /** Create a RouteGuide server listening on {@code port} using {@code featureFile} database. */
  44.   public RouteGuideServer(int port, URL featureFile) throws IOException {
  45.     this(ServerBuilder.forPort(port), port, RouteGuideUtil.parseFeatures(featureFile));
  46.   }
  47.  
  48.   /** Create a RouteGuide server using serverBuilder as a base and features as data. */
  49.   public RouteGuideServer(ServerBuilder<?> serverBuilder, int port, Collection<Feature> features) {
  50.     this.port = port;
  51.     server = serverBuilder.addService(new RouteGuideService(features))
  52.         .build();
  53.   }
  54.  
  55.   /** Start serving requests. */
  56.   public void start() throws IOException {
  57.     server.start();
  58.     logger.info("Server started, listening on " + port);
  59.     Runtime.getRuntime().addShutdownHook(new Thread() {
  60.       @Override
  61.       public void run() {
  62.         // Use stderr here since the logger may has been reset by its JVM shutdown hook.
  63.         System.err.println("*** shutting down gRPC server since JVM is shutting down");
  64.         RouteGuideServer.this.stop();
  65.         System.err.println("*** server shut down");
  66.       }
  67.     });
  68.   }
  69.  
  70.   /** Stop serving requests and shutdown resources. */
  71.   public void stop() {
  72.     if (server != null) {
  73.       server.shutdown();
  74.     }
  75.   }
  76.  
  77.   /**
  78.    * Await termination on the main thread since the grpc library uses daemon threads.
  79.    */
  80.   private void blockUntilShutdown() throws InterruptedException {
  81.     if (server != null) {
  82.       server.awaitTermination();
  83.     }
  84.   }
  85.  
  86.   /**
  87.    * Main method.  This comment makes the linter happy.
  88.    */
  89.   public static void main(String[] args) throws Exception {
  90.     RouteGuideServer server = new RouteGuideServer(8980);
  91.     server.start();
  92.     server.blockUntilShutdown();
  93.   }
  94.  
  95.   /**
  96.    * Our implementation of RouteGuide service.
  97.    *
  98.    * <p>See route_guide.proto for details of the methods.
  99.    */
  100.   private static class RouteGuideService extends RouteGuideGrpc.RouteGuideImplBase {
  101.     private final Collection<Feature> features;
  102.     private final ConcurrentMap<Point, List<RouteNote>> routeNotes =
  103.         new ConcurrentHashMap<Point, List<RouteNote>>();
  104.  
  105.     RouteGuideService(Collection<Feature> features) {
  106.       this.features = features;
  107.     }
  108.  
  109.     /**
  110.      * Gets the {@link Feature} at the requested {@link Point}. If no feature at that location
  111.      * exists, an unnamed feature is returned at the provided location.
  112.      *
  113.      * @param request the requested location for the feature.
  114.      * @param responseObserver the observer that will receive the feature at the requested point.
  115.      */
  116.     @Override
  117.     public void getFeature(Point request, StreamObserver<Feature> responseObserver) {
  118.       responseObserver.onNext(checkFeature(request));
  119.       responseObserver.onCompleted();
  120.     }
  121.  
  122.     /**
  123.      * Gets all features contained within the given bounding {@link Rectangle}.
  124.      *
  125.      * @param request the bounding rectangle for the requested features.
  126.      * @param responseObserver the observer that will receive the features.
  127.      */
  128.     @Override
  129.     public void listFeatures(Rectangle request, StreamObserver<Feature> responseObserver) {
  130.       int left = min(request.getLo().getLongitude(), request.getHi().getLongitude());
  131.       int right = max(request.getLo().getLongitude(), request.getHi().getLongitude());
  132.       int top = max(request.getLo().getLatitude(), request.getHi().getLatitude());
  133.       int bottom = min(request.getLo().getLatitude(), request.getHi().getLatitude());
  134.  
  135.       for (Feature feature : features) {
  136.         if (!RouteGuideUtil.exists(feature)) {
  137.           continue;
  138.         }
  139.  
  140.         int lat = feature.getLocation().getLatitude();
  141.         int lon = feature.getLocation().getLongitude();
  142.         if (lon >= left && lon <= right && lat >= bottom && lat <= top) {
  143.           responseObserver.onNext(feature);
  144.         }
  145.       }
  146.       responseObserver.onCompleted();
  147.     }
  148.  
  149.     /**
  150.      * Gets a stream of points, and responds with statistics about the "trip": number of points,
  151.      * number of known features visited, total distance traveled, and total time spent.
  152.      *
  153.      * @param responseObserver an observer to receive the response summary.
  154.      * @return an observer to receive the requested route points.
  155.      */
  156.     @Override
  157.     public StreamObserver<Point> recordRoute(final StreamObserver<RouteSummary> responseObserver) {
  158.       return new StreamObserver<Point>() {
  159.         int pointCount;
  160.         int featureCount;
  161.         int distance;
  162.         Point previous;
  163.         final long startTime = System.nanoTime();
  164.  
  165.         @Override
  166.         public void onNext(Point point) {
  167.           pointCount++;
  168.           if (RouteGuideUtil.exists(checkFeature(point))) {
  169.             featureCount++;
  170.           }
  171.           // For each point after the first, add the incremental distance from the previous point to
  172.           // the total distance value.
  173.           if (previous != null) {
  174.             distance += calcDistance(previous, point);
  175.           }
  176.           previous = point;
  177.         }
  178.  
  179.         @Override
  180.         public void onError(Throwable t) {
  181.           logger.log(Level.WARNING, "recordRoute cancelled");
  182.         }
  183.  
  184.         @Override
  185.         public void onCompleted() {
  186.           long seconds = NANOSECONDS.toSeconds(System.nanoTime() - startTime);
  187.           responseObserver.onNext(RouteSummary.newBuilder().setPointCount(pointCount)
  188.               .setFeatureCount(featureCount).setDistance(distance)
  189.               .setElapsedTime((int) seconds).build());
  190.           responseObserver.onCompleted();
  191.         }
  192.       };
  193.     }
  194.  
  195.     /**
  196.      * Receives a stream of message/location pairs, and responds with a stream of all previous
  197.      * messages at each of those locations.
  198.      *
  199.      * @param responseObserver an observer to receive the stream of previous messages.
  200.      * @return an observer to handle requested message/location pairs.
  201.      */
  202.     @Override
  203.     public StreamObserver<RouteNote> routeChat(final StreamObserver<RouteNote> responseObserver) {
  204.       return new StreamObserver<RouteNote>() {
  205.         @Override
  206.         public void onNext(RouteNote note) {
  207.           List<RouteNote> notes = getOrCreateNotes(note.getLocation());
  208.  
  209.           // Respond with all previous notes at this location.
  210.           for (RouteNote prevNote : notes.toArray(new RouteNote[0])) {
  211.             responseObserver.onNext(prevNote);
  212.           }
  213.  
  214.           // Now add the new note to the list
  215.           notes.add(note);
  216.         }
  217.  
  218.         @Override
  219.         public void onError(Throwable t) {
  220.           logger.log(Level.WARNING, "routeChat cancelled");
  221.         }
  222.  
  223.         @Override
  224.         public void onCompleted() {
  225.           responseObserver.onCompleted();
  226.         }
  227.       };
  228.     }
  229.  
  230.     /**
  231.      * Get the notes list for the given location. If missing, create it.
  232.      */
  233.     private List<RouteNote> getOrCreateNotes(Point location) {
  234.       List<RouteNote> notes = Collections.synchronizedList(new ArrayList<RouteNote>());
  235.       List<RouteNote> prevNotes = routeNotes.putIfAbsent(location, notes);
  236.       return prevNotes != null ? prevNotes : notes;
  237.     }
  238.  
  239.     /**
  240.      * Gets the feature at the given point.
  241.      *
  242.      * @param location the location to check.
  243.      * @return The feature object at the point. Note that an empty name indicates no feature.
  244.      */
  245.     private Feature checkFeature(Point location) {
  246.       for (Feature feature : features) {
  247.         if (feature.getLocation().getLatitude() == location.getLatitude()
  248.             && feature.getLocation().getLongitude() == location.getLongitude()) {
  249.           return feature;
  250.         }
  251.       }
  252.  
  253.       // No feature was found, return an unnamed feature.
  254.       return Feature.newBuilder().setName("").setLocation(location).build();
  255.     }
  256.  
  257.     /**
  258.      * Calculate the distance between two points using the "haversine" formula.
  259.      * The formula is based on http://mathforum.org/library/drmath/view/51879.html.
  260.      *
  261.      * @param start The starting point
  262.      * @param end The end point
  263.      * @return The distance between the points in meters
  264.      */
  265.     private static int calcDistance(Point start, Point end) {
  266.       int r = 6371000; // earth radius in meters
  267.       double lat1 = toRadians(RouteGuideUtil.getLatitude(start));
  268.       double lat2 = toRadians(RouteGuideUtil.getLatitude(end));
  269.       double lon1 = toRadians(RouteGuideUtil.getLongitude(start));
  270.       double lon2 = toRadians(RouteGuideUtil.getLongitude(end));
  271.       double deltaLat = lat2 - lat1;
  272.       double deltaLon = lon2 - lon1;
  273.  
  274.       double a = sin(deltaLat / 2) * sin(deltaLat / 2)
  275.           + cos(lat1) * cos(lat2) * sin(deltaLon / 2) * sin(deltaLon / 2);
  276.       double c = 2 * atan2(sqrt(a), sqrt(1 - a));
  277.  
  278.       return (int) (r * c);
  279.     }
  280.   }
  281. }


客户端streaming demo如下：

  1. import com.google.common.annotations.VisibleForTesting;
  2. import com.google.protobuf.Message;
  3. import io.grpc.ManagedChannel;
  4. import io.grpc.ManagedChannelBuilder;
  5. import io.grpc.Status;
  6. import io.grpc.StatusRuntimeException;
  7. import io.grpc.examples.routeguide.Feature;
  8. import io.grpc.examples.routeguide.Point;
  9. import io.grpc.examples.routeguide.Rectangle;
  10. import io.grpc.examples.routeguide.RouteGuideGrpc;
  11. import io.grpc.examples.routeguide.RouteGuideGrpc.RouteGuideBlockingStub;
  12. import io.grpc.examples.routeguide.RouteGuideGrpc.RouteGuideStub;
  13. import io.grpc.examples.routeguide.RouteNote;
  14. import io.grpc.examples.routeguide.RouteSummary;
  15. import io.grpc.stub.StreamObserver;
  16. import java.io.IOException;
  17. import java.util.Iterator;
  18. import java.util.List;
  19. import java.util.Random;
  20. import java.util.concurrent.CountDownLatch;
  21. import java.util.concurrent.TimeUnit;
  22. import java.util.logging.Level;
  23. import java.util.logging.Logger;
  24.  
  25. /**
  26.  * Sample client code that makes gRPC calls to the server.
  27.  */
  28. public class RouteGuideClient {
  29.   private static final Logger logger = Logger.getLogger(RouteGuideClient.class.getName());
  30.  
  31.   private final ManagedChannel channel;
  32.   private final RouteGuideBlockingStub blockingStub;
  33.   private final RouteGuideStub asyncStub;
  34.  
  35.   private Random random = new Random();
  36.   private TestHelper testHelper;
  37.  
  38.   /** Construct client for accessing RouteGuide server at {@code host:port}. */
  39.   public RouteGuideClient(String host, int port) {
  40.     this(ManagedChannelBuilder.forAddress(host, port).usePlaintext());
  41.   }
  42.  
  43.   /** Construct client for accessing RouteGuide server using the existing channel. */
  44.   public RouteGuideClient(ManagedChannelBuilder<?> channelBuilder) {
  45.     channel = channelBuilder.build();
  46.     blockingStub = RouteGuideGrpc.newBlockingStub(channel);
  47.     asyncStub = RouteGuideGrpc.newStub(channel);
  48.   }
  49.  
  50.   public void shutdown() throws InterruptedException {
  51.     channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
  52.   }
  53.  
  54.   /**
  55.    * Blocking unary call example.  Calls getFeature and prints the response.
  56.    */
  57.   public void getFeature(int lat, int lon) {
  58.     info("*** GetFeature: lat={0} lon={1}", lat, lon);
  59.  
  60.     Point request = Point.newBuilder().setLatitude(lat).setLongitude(lon).build();
  61.  
  62.     Feature feature;
  63.     try {
  64.       feature = blockingStub.getFeature(request);
  65.       if (testHelper != null) {
  66.         testHelper.onMessage(feature);
  67.       }
  68.     } catch (StatusRuntimeException e) {
  69.       warning("RPC failed: {0}", e.getStatus());
  70.       if (testHelper != null) {
  71.         testHelper.onRpcError(e);
  72.       }
  73.       return;
  74.     }
  75.     if (RouteGuideUtil.exists(feature)) {
  76.       info("Found feature called \"{0}\" at {1}, {2}",
  77.           feature.getName(),
  78.           RouteGuideUtil.getLatitude(feature.getLocation()),
  79.           RouteGuideUtil.getLongitude(feature.getLocation()));
  80.     } else {
  81.       info("Found no feature at {0}, {1}",
  82.           RouteGuideUtil.getLatitude(feature.getLocation()),
  83.           RouteGuideUtil.getLongitude(feature.getLocation()));
  84.     }
  85.   }
  86.  
  87.   /**
  88.    * Blocking server-streaming example. Calls listFeatures with a rectangle of interest. Prints each
  89.    * response feature as it arrives.
  90.    */
  91.   public void listFeatures(int lowLat, int lowLon, int hiLat, int hiLon) {
  92.     info("*** ListFeatures: lowLat={0} lowLon={1} hiLat={2} hiLon={3}", lowLat, lowLon, hiLat,
  93.         hiLon);
  94.  
  95.     Rectangle request =
  96.         Rectangle.newBuilder()
  97.             .setLo(Point.newBuilder().setLatitude(lowLat).setLongitude(lowLon).build())
  98.             .setHi(Point.newBuilder().setLatitude(hiLat).setLongitude(hiLon).build()).build();
  99.     Iterator<Feature> features;
  100.     try {
  101.       features = blockingStub.listFeatures(request);
  102.       for (int i = 1; features.hasNext(); i++) {
  103.         Feature feature = features.next();
  104.         info("Result #" + i + ": {0}", feature);
  105.         if (testHelper != null) {
  106.           testHelper.onMessage(feature);
  107.         }
  108.       }
  109.     } catch (StatusRuntimeException e) {
  110.       warning("RPC failed: {0}", e.getStatus());
  111.       if (testHelper != null) {
  112.         testHelper.onRpcError(e);
  113.       }
  114.     }
  115.   }
  116.  
  117.   /**
  118.    * Async client-streaming example. Sends {@code numPoints} randomly chosen points from {@code
  119.    * features} with a variable delay in between. Prints the statistics when they are sent from the
  120.    * server.
  121.    */
  122.   public void recordRoute(List<Feature> features, int numPoints) throws InterruptedException {
  123.     info("*** RecordRoute");
  124.     final CountDownLatch finishLatch = new CountDownLatch(1);
  125.     StreamObserver<RouteSummary> responseObserver = new StreamObserver<RouteSummary>() {
  126.       @Override
  127.       public void onNext(RouteSummary summary) {
  128.         info("Finished trip with {0} points. Passed {1} features. "
  129.             + "Travelled {2} meters. It took {3} seconds.", summary.getPointCount(),
  130.             summary.getFeatureCount(), summary.getDistance(), summary.getElapsedTime());
  131.         if (testHelper != null) {
  132.           testHelper.onMessage(summary);
  133.         }
  134.       }
  135.  
  136.       @Override
  137.       public void onError(Throwable t) {
  138.         warning("RecordRoute Failed: {0}", Status.fromThrowable(t));
  139.         if (testHelper != null) {
  140.           testHelper.onRpcError(t);
  141.         }
  142.         finishLatch.countDown();
  143.       }
  144.  
  145.       @Override
  146.       public void onCompleted() {
  147.         info("Finished RecordRoute");
  148.         finishLatch.countDown();
  149.       }
  150.     };
  151.  
  152.     StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);
  153.     try {
  154.       // Send numPoints points randomly selected from the features list.
  155.       for (int i = 0; i < numPoints; ++i) {
  156.         int index = random.nextInt(features.size());
  157.         Point point = features.get(index).getLocation();
  158.         info("Visiting point {0}, {1}", RouteGuideUtil.getLatitude(point),
  159.             RouteGuideUtil.getLongitude(point));
  160.         requestObserver.onNext(point);
  161.         // Sleep for a bit before sending the next one.
  162.         Thread.sleep(random.nextInt(1000) + 500);
  163.         if (finishLatch.getCount() == 0) {
  164.           // RPC completed or errored before we finished sending.
  165.           // Sending further requests won't error, but they will just be thrown away.
  166.           return;
  167.         }
  168.       }
  169.     } catch (RuntimeException e) {
  170.       // Cancel RPC
  171.       requestObserver.onError(e);
  172.       throw e;
  173.     }
  174.     // Mark the end of requests
  175.     requestObserver.onCompleted();
  176.  
  177.     // Receiving happens asynchronously
  178.     if (!finishLatch.await(1, TimeUnit.MINUTES)) {
  179.       warning("recordRoute can not finish within 1 minutes");
  180.     }
  181.   }
  182.  
  183.   /**
  184.    * Bi-directional example, which can only be asynchronous. Send some chat messages, and print any
  185.    * chat messages that are sent from the server.
  186.    */
  187.   public CountDownLatch routeChat() {
  188.     info("*** RouteChat");
  189.     final CountDownLatch finishLatch = new CountDownLatch(1);
  190.     StreamObserver<RouteNote> requestObserver =
  191.         asyncStub.routeChat(new StreamObserver<RouteNote>() {
  192.           @Override
  193.           public void onNext(RouteNote note) {
  194.             info("Got message \"{0}\" at {1}, {2}", note.getMessage(), note.getLocation()
  195.                 .getLatitude(), note.getLocation().getLongitude());
  196.             if (testHelper != null) {
  197.               testHelper.onMessage(note);
  198.             }
  199.           }
  200.  
  201.           @Override
  202.           public void onError(Throwable t) {
  203.             warning("RouteChat Failed: {0}", Status.fromThrowable(t));
  204.             if (testHelper != null) {
  205.               testHelper.onRpcError(t);
  206.             }
  207.             finishLatch.countDown();
  208.           }
  209.  
  210.           @Override
  211.           public void onCompleted() {
  212.             info("Finished RouteChat");
  213.             finishLatch.countDown();
  214.           }
  215.         });
  216.  
  217.     try {
  218.       RouteNote[] requests =
  219.           {newNote("First message", 0, 0), newNote("Second message", 0, 1),
  220.               newNote("Third message", 1, 0), newNote("Fourth message", 1, 1)};
  221.  
  222.       for (RouteNote request : requests) {
  223.         info("Sending message \"{0}\" at {1}, {2}", request.getMessage(), request.getLocation()
  224.             .getLatitude(), request.getLocation().getLongitude());
  225.         requestObserver.onNext(request);
  226.       }
  227.     } catch (RuntimeException e) {
  228.       // Cancel RPC
  229.       requestObserver.onError(e);
  230.       throw e;
  231.     }
  232.     // Mark the end of requests
  233.     requestObserver.onCompleted();
  234.  
  235.     // return the latch while receiving happens asynchronously
  236.     return finishLatch;
  237.   }
  238.  
  239.   /** Issues several different requests and then exits. */
  240.   public static void main(String[] args) throws InterruptedException {
  241.     List<Feature> features;
  242.     try {
  243.       features = RouteGuideUtil.parseFeatures(RouteGuideUtil.getDefaultFeaturesFile());
  244.     } catch (IOException ex) {
  245.       ex.printStackTrace();
  246.       return;
  247.     }
  248.  
  249.     RouteGuideClient client = new RouteGuideClient("localhost", 8980);
  250.     try {
  251.       // Looking for a valid feature
  252.       client.getFeature(409146138, -746188906);
  253.  
  254.       // Feature missing.
  255.       client.getFeature(0, 0);
  256.  
  257.       // Looking for features between 40, -75 and 42, -73.
  258.       client.listFeatures(400000000, -750000000, 420000000, -730000000);
  259.  
  260.       // Record a few randomly selected points from the features file.
  261.       client.recordRoute(features, 10);
  262.  
  263.       // Send and receive some notes.
  264.       CountDownLatch finishLatch = client.routeChat();
  265.  
  266.       if (!finishLatch.await(1, TimeUnit.MINUTES)) {
  267.         client.warning("routeChat can not finish within 1 minutes");
  268.       }
  269.     } finally {
  270.       client.shutdown();
  271.     }
  272.   }
  273.  
  274.   private void info(String msg, Object... params) {
  275.     logger.log(Level.INFO, msg, params);
  276.   }
  277.  
  278.   private void warning(String msg, Object... params) {
  279.     logger.log(Level.WARNING, msg, params);
  280.   }
  281.  
  282.   private RouteNote newNote(String message, int lat, int lon) {
  283.     return RouteNote.newBuilder().setMessage(message)
  284.         .setLocation(Point.newBuilder().setLatitude(lat).setLongitude(lon).build()).build();
  285.   }
  286.  
  287.   /**
  288.    * Only used for unit test, as we do not want to introduce randomness in unit test.
  289.    */
  290.   @VisibleForTesting
  291.   void setRandom(Random random) {
  292.     this.random = random;
  293.   }
  294.  
  295.   /**
  296.    * Only used for helping unit test.
  297.    */
  298.   @VisibleForTesting
  299.   interface TestHelper {
  300.     /**
  301.      * Used for verify/inspect message received from server.
  302.      */
  303.     void onMessage(Message message);
  304.  
  305.     /**
  306.      * Used for verify/inspect error received from server.
  307.      */
  308.     void onRpcError(Throwable exception);
  309.   }
  310.  
  311.   @VisibleForTesting
  312.   void setTestHelper(TestHelper testHelper) {
  313.     this.testHelper = testHelper;
  314.   }
  315. }

服务端和客户端双向 streaming，就是上面两个结合了


2 总结

  gRPC 服务调用支持同步和异步方式，同时也支持普通的 RPC 和 streaming 模式，可以最大程度满足业务的需求。 对于 streaming 模式，可以充分利用 HTTP/2.0 协议的多路复用功能，实现在一条 HTTP 链路上并行双向传输数据，有效的解决了 HTTP/1.X 的数据单向传输问题，在大幅减少 HTTP 连接的情况下，充分利用单条链路的性能，可以媲美传统的 RPC 私有长连接协议：更少的链路、更高的性能。

  gRPC 的网络 I/O 通信基于 Netty 构建，服务调用底层统一使用异步方式，同步调用是在异步的基础上做了上层封装。因此，gRPC 的异步化是比较彻底的，对于提升 I/O 密集型业务的吞吐量和可靠性有很大的帮助。


 下一章我会为大家讲解grpc关于SSL部分。



声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/382442
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号