Trong mô hình gRPC Bidirectional Streaming, Client và Server có thể trao đổi nhiều request và response qua một kết nối TCP duy nhất. Các request và response này có thể hoàn toàn độc lập với nhau. Client và Server sẽ đóng cuộc gọi khi chúng hoàn tất việc nghiệp vụ.
Hãy xem một số ví dụ ứng dụng streaming ở 2 phía Client và Server.
- Màn hình tìm kiếm của Google: Ngay sau khi chúng ta nhập một từ khóa, nó sẽ được gửi đến máy chủ và máy chủ ngay lập tức phản hồi với các từ khóa tìm kiếm có thể.
- Netflix / YouTube: Dựa trên các video chúng ta tìm kiếm/xem, chúng ta sẽ có thêm đề xuất liên quan đến các video đó.
Sample Application
Trong ví dụ này, chúng ta sẽ tạo ra một ứng dụng GPS cho ô tô. Ví dụ đi từ điểm A đến điểm B cách nhau 100m. Khi bắt đầu lái xe, chúng ta sẽ đồng bộ với máy chủ sau mỗi 3 giây và theo dõi vị trí của chúng ta. Máy chủ sẽ phản hồi khoảng cách còn lại đến điểm đến B và thời gian ước tính cần thiết để đến đích cho đến khi chuyến đi hoàn thành.
Protobuf – Service Definition
Khi biết nghiệp vụ, chúng ta sẽ định nghĩa service xử lý nghiệp vụ cho chúng. Phương thức navigate được implements ở phía Server nhận kiểu dữ liệu đầu vào và trả kiểu dữ liệu đầu ra theo như mong đợi. Chúng ta sử dụng từ khóa stream ở cả request và response để chỉ ra rằng đó sẽ là dịch vụ streaming ở cả 2 chiều.
syntax = "proto3";
package gps;
option java_package = "example.gps";
option java_multiple_files = true;
message TripRequest {
int32 distanceTravelled = 1;
}
message TripResponse {
int32 remainingDistance = 1;
int32 timeToDestination = 2;
}
service NavigationService {
// grpc bidirectional stream
rpc navigate(stream TripRequest) returns (stream TripResponse);
}
Khi chúng ta chạy lệnh maven dưới đây, maven sẽ tự động tạo code cho client application và server application bằng công cụ protoc.
mvn clean compile
Class NavigatorServiceImplBase là abstract class được tạo tự động khi gen code cần được phía Server implements. Tương tự NavigatorServiceStub là class mà phía client application sử dụng để gửi yêu cầu đến server.
gRPC Bidirectional Streaming – Server Side
Service Implementation: class này kế thừa abstract class NavigatorServiceImplBase để implement phương thức navigate (triển khai nghiệp vụ) và phản hồi lại cho request gọi phương thức placeOrder ở phía Client. Client sẽ gửi nhiều request đến Server và nhận về nhiều response. Khi Client gọi phương thức onCompleted, nó sẽ thông báo cho Server rằng nó đã đến đích. Lúc đó Server cũng có thể kết thúc cuộc gọi.
- Khi bắt đầu, chúng ta khởi tạo khoảng cách là 100m, chúng ta khởi tạo một con số bất kỳ là thời gian bắt đầu.
- Cứ sau 3 giây, Client sẽ gửi khoảng cách mà nó đã di chuyển được trong 3 giây trong lệnh gọi phương thức
onNext. - Khi Server nhận được cuộc gọi
onNexttừ Client, Server sẽ kiểm tra tổng số quãng đường khách hàng đã đi và tính quãng đường còn lại. - Chúng ta cũng biết tốc độ của khách hàng (Quãng đường đã đi / Thời gian di chuyển). Sử dụng tốc độ này, chúng ta cũng tính toán khoảng thời gian cần thiết để đến đích.
public class TripRequestObserver implements StreamObserver<TripRequest> {
private final int totalDistance = 100;
private LocalTime startTime = LocalTime.now();
private int distanceTraveled;
private final StreamObserver<TripResponse> tripResponseStreamObserver;
public TripRequestObserver(StreamObserver<TripResponse> tripResponseStreamObserver) {
this.tripResponseStreamObserver = tripResponseStreamObserver;
}
@Override
public void onNext(TripRequest tripRequest) {
this.distanceTraveled = Math.min(totalDistance, (this.distanceTraveled + tripRequest.getDistanceTravelled()));
int remainingDistance = Math.max(0, (totalDistance - distanceTraveled));
// the client has reached destination
if(remainingDistance == 0){
this.tripResponseStreamObserver.onNext(TripResponse.getDefaultInstance());
return;
}
// client has not yet reached destination
long elapsedDuration = Duration.between(this.startTime, LocalTime.now()).getSeconds();
elapsedDuration = elapsedDuration < 1 ? 1 : elapsedDuration;
double currentSpeed = (distanceTraveled * 1.0d) / elapsedDuration;
int timeToReach = (int) (remainingDistance / currentSpeed);
TripResponse tripResponse = TripResponse.newBuilder()
.setRemainingDistance(remainingDistance)
.setTimeToDestination(timeToReach)
.build();
this.tripResponseStreamObserver.onNext(tripResponse);
}
@Override
public void onError(Throwable throwable) {
System.out.println("=========== Client onError ===========");
}
@Override
public void onCompleted() {
System.out.println("=========== Client onCompleted ===========");
this.tripResponseStreamObserver.onCompleted();
System.out.println("Client reached safely");
}
}
Việc triển khai ở trên là logic nghiệp vụ thực tế cần được thực thi bất cứ khi nào Client gửi vị trí hiện tại đến Server.
public class NavigationService extends NavigationServiceGrpc.NavigationServiceImplBase {
@Override
public StreamObserver<TripRequest> navigate(StreamObserver<TripResponse> responseObserver) {
return new TripRequestObserver(responseObserver);
}
}
Sau khi implements xong, chúng ta cần start gRPC server để cung cấp dịch vụ cho Client.
public class GPSServer {
public static void main(String[] args) throws IOException, InterruptedException {
// build gRPC server
Server server = ServerBuilder.forPort(6565)
.addService(new NavigationService())
.build();
// start
server.start();
// shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
System.out.println("GPS is shutting down!");
server.shutdown();
}));
server.awaitTermination();
}
}
Bây giờ thì gRPC server đã sẵn sàng. Chúng ta sẽ qua Client Side.
gRPC Bidirectional Streaming – Client Side
Protobuf đã gen code cho phía Client application. Trên phía Client chúng ta cần thực hiện các bước sau để gửi request và nhận lại response. Bước đầu tiên để thực hiện gửi request, chúng ta cần có một implementation của StreamObserver.
- Cứ mỗi 3 giây di chuyển, Client sẽ gửi khoảng cách di chuyển được đến Server. Client tiếp tục làm điều này lặp đi lặp lại cho đến khi Client đến đích.
- Khi Client đến đích, chúng ta sẽ cho Server biết rằng cuộc gọi này có thể kết thúc.
public class TripResponseStreamObserver implements StreamObserver<TripResponse> {
private StreamObserver<TripRequest> requestStreamObserver;
@Override
public void onNext(TripResponse tripResponse) {
if(tripResponse.getRemainingDistance() > 0){
print(tripResponse);
this.drive();
}else{
this.requestStreamObserver.onCompleted();
}
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onCompleted() {
System.out.println("Trip Completed");
}
public void startTrip(StreamObserver<TripRequest> requestStreamObserver){
this.requestStreamObserver = requestStreamObserver;
this.drive();
}
private void drive(){
Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
TripRequest tripRequest = TripRequest.newBuilder().setDistanceTravelled(ThreadLocalRandom.current().nextInt(1, 10)).build();
requestStreamObserver.onNext(tripRequest);
}
private void print(TripResponse tripResponse){
System.out.println(LocalTime.now() + ": Remaining Distance : " + tripResponse.getRemainingDistance());
System.out.println(LocalTime.now() + ": Time To Reach (sec): " + tripResponse.getTimeToDestination());
System.out.println("------------------------------");
}
}
Tiếp theo là tạo kết nối và gửi request:
Tạo kênh (create channel): client apllication phải tạo một kênh kết nối với gRPC server.
Sử dụng Stub: Trong trường hợp này Client apllication sẽ sử dụng non-blocking stub để thực hiện truyền tham số và gửi request.
Trong ví dụ này chúng ta sẽ tạo một class JUnit để hoạt động như một gRPC client. Hãy lưu ý rằng client application (hay gRPC client) có thể là bất cứ ngôn ngữ gì. Nó thậm chí có thể là một microservice khác.
public class BiDirectionalStreamingTest {
private ManagedChannel channel;
private NavigationServiceGrpc.NavigationServiceStub clientStub;
@Before
public void setup(){
this.channel = ManagedChannelBuilder.forAddress("localhost", 6565)
.usePlaintext()
.build();
this.clientStub = NavigationServiceGrpc.newStub(channel);
}
@Test
public void tripTest() throws InterruptedException {
TripResponseStreamObserver tripResponseStreamObserver = new TripResponseStreamObserver();
StreamObserver<TripRequest> requestStreamObserver = this.clientStub.navigate(tripResponseStreamObserver);
tripResponseStreamObserver.startTrip(requestStreamObserver);
}
@After
public void teardown(){
this.channel.shutdown();
}
}
Server Output:
10:19:05.136686: Remaining Distance : 98
10:19:05.136806: Time To Reach (sec): 98
------------------------------
10:19:08.146316: Remaining Distance : 91
10:19:08.146579: Time To Reach (sec): 50
------------------------------
10:19:11.152504: Remaining Distance : 82
10:19:11.152759: Time To Reach (sec): 36
...
10:19:59.244266: Remaining Distance : 8
10:19:59.244463: Time To Reach (sec): 4
------------------------------
10:20:02.249562: Remaining Distance : 7
10:20:02.249779: Time To Reach (sec): 4
------------------------------
Trip Completed
Client output:
=========== Client onCompleted ===========
Client reached safely
Tổng kết
Vậy là chúng ta vừa làm quen cách tạo một ứng dụng gRPC bidirectional Streaming. Hi vọng bài viết hữu ích với mọi người.
Nguồn: https://thenewstack.wordpress.com/2021/11/24/grpc-grpc-bidirectional-streaming/
Follow me: thenewstack.wordpress.com