Command Palette
Search for a command to run...
Comments
Join the discussionNo comments yet. Be the first to comment.
Microservice java
Part 1 of 1
Search for a command to run...
No comments yet. Be the first to comment.
Trong gRPC, file .proto đóng vai trò như hợp đồng (contract) giữa client và server.
proto là gì?Đây là nơi định nghĩa:
Ví dụ:
service UserService {
rpc GetUser (UserRequest) returns (UserResponse);
}
Thay vì viết code thủ công, gRPC sử dụng protoc để generate:
Trong java ta dùng các thư viện để gen code từ proto
protobuf-maven-plugin – để chạy protocprotoc-gen-grpc-java – để generate gRPC stubCòn trong spring thì ta dùng thêm net.devh:grpc-spring-boot-starter để tích hợp gRPC với Spring Boot, giúp:
Cụ thể ta có thể cấu hình plugin maven như sau:
<properties>
<java.version>17</java.version>
<grpc.version>1.61.0</grpc.version>
<protobuf.version>3.25.3</protobuf.version>
</properties>
<dependencies>
<!-- gRPC Spring Boot Starter -->
<dependency>
<groupId>net.devh</groupId>
<artifactId>grpc-spring-boot-starter</artifactId>
<version>2.15.0.RELEASE</version>
</dependency>
<!-- Protobuf -->
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>${protobuf.version}</version>
</dependency>
<!-- gRPC -->
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
<version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
<version>${grpc.version}</version>
</dependency>
</dependencies>
<build>
<extensions>
<!-- Detect OS để tải đúng binary protoc -->
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.7.1</version>
</extension>
</extensions>
<plugins>
<!-- Plugin generate code từ proto -->
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
<version>0.6.1</version>
<configuration>
<protocArtifact>
com.google.protobuf:protoc:\({protobuf.version}:exe:\){os.detected.classifier}
</protocArtifact>
<pluginId>grpc-java</pluginId>
<pluginArtifact>
io.grpc:protoc-gen-grpc-java:\({grpc.version}:exe:\){os.detected.classifier}
</pluginArtifact>
</configuration>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>compile-custom</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
Ví dụ: với proto sau:
syntax = "proto3";
package auth;
option java_multiple_files = true;
option java_package = "com.aqbtech.common.proto.auth";
service AuthService {
rpc Login (LoginRequest) returns (LoginResponse);
}
message LoginRequest {
string username = 1;
string password = 2;
}
message LoginResponse {
string access_token = 1;
string refresh_token = 2;
}
Sau khi chạy protoc, ta sẽ có:
LoginRequest, LoginResponse class đại diện cho dữ liệu (immutable, builder pattern)AuthServiceGrpcchứa:AuthServiceImplBase (server implement)AuthServiceBlockingStubAuthServiceStubAuthServiceFutureStubTa có 1 service implement như sau:
public class AuthServiceImpl extends AuthServiceGrpc.AuthServiceImplBase {
@Override
public void login(LoginRequest request,
StreamObserver<LoginResponse> responseObserver) {
LoginResponse response = LoginResponse.newBuilder()
.setAccessToken("token")
.setRefreshToken("refresh")
.build();
responseObserver.onNext(response);
responseObserver.onCompleted();
}
}
Trong đó StreamObserver là một interface trong gRPC Java, dùng để nhận hoặc gửi dữ liệu theo dạng stream giữa client và server.
Interface này có 3 phương thức chính:
publicinterfaceStreamObserver<V> {
void onNext(V value);
void onError(Throwable t);
void onCompleted();
}
onNext(V value)Gửi hoặc nhận một messageonError(Throwable t)Báo lỗi và kết thúc streamonCompleted()Kết thúc stream thành côngListenableFutureBlocking stub là stub gọi đồng bộ (synchronous), nghĩa là:
Ví dụ:
LoginResponse response = blockingStub.login(request);
Đặc điểm: Không phù hợp với: High throughput system, UI thread
Async stub là stub gọi bất đồng bộ (non-blocking):
StreamObserver)Callback là một cơ chế trong lập trình bất đồng bộ, trong đó:
Trong ngữ cảnh gRPC Async stub, khi gọi async:
asyncStub.login(request, new StreamObserver<LoginResponse>() {
@Override
public void onNext(LoginResponse value) {
// xử lý khi nhận được response
}
@Override
public void onError(Throwable t) {
// xử lý khi có lỗi
}
@Override
public void onCompleted() {
// xử lý khi kết thúc
}
});
Thì StreamObserver chính là callback object.
Cách hoạt động
onNext()onError()onCompleted()Đặc điểm của callback
Future stub là dạng trung gian giữa blocking và async:
ListenableFutureget())Ví dụ:
ListenableFuture<LoginResponse> future = futureStub.login(request);
LoginResponse response = future.get();
Đặc điểm:
Trong gRPC, cách client gửi dữ liệu phụ thuộc vào kiểu RPC được định nghĩa trong .proto. Có các kiểu gửi (RPC types) trong gRPC, định nghĩa 4 kiểu giao tiếp chính giữa client và server:
Proto syntax:
rpc GetUser (UserRequest) returns (UserResponse);
Proto syntax:
rpc GetOrders (OrderRequest) returns (stream OrderResponse);
Proto syntax:
rpc UploadLogs (stream LogRequest) returns (LogResponse);
Proto syntax:
rpc Chat (stream ChatMessage) returns (stream ChatMessage);
Trong gRPC, việc xử lý lỗi không sử dụng HTTP status như REST, mà dựa trên một hệ thống status code chuẩn hoá được định nghĩa sẵn. Điều này giúp đảm bảo tính nhất quán giữa các service, bất kể ngôn ngữ hay nền tảng.
Status code trong gRPC là mã trạng thái được server trả về để mô tả kết quả của một RPC call.
Mỗi response trong gRPC đều đi kèm:
Cấu trúc logic:
status = {
code: StatusCode,
description: String,
metadata: Metadata (optional)
}
gRPC định nghĩa một tập hợp status code chuẩn (theo io.grpc.Status trong Java). Các code quan trọng:
OK (0)
Request xử lý thành công.
INVALID_ARGUMENT
Dữ liệu đầu vào không hợp lệ (validation fail)
NOT_FOUND
Không tìm thấy resource
ALREADY_EXISTS
Resource đã tồn tại
FAILED_PRECONDITION
Trạng thái hệ thống không phù hợp để thực hiện request
OUT_OF_RANGE
Giá trị nằm ngoài phạm vi cho phép
UNAUTHENTICATED
Chưa xác thực (thiếu/invalid token)
PERMISSION_DENIED
Không có quyền truy cập
INTERNAL
Lỗi nội bộ server
UNAVAILABLE
Service không sẵn sàng (down, timeout, network)
DEADLINE_EXCEEDED
Request bị timeout
RESOURCE_EXHAUSTED
Hết tài nguyên (rate limit, memory, quota)
ABORTED
Operation bị huỷ do conflict
CANCELLED
Client chủ động huỷ request
UNKNOWN
Lỗi không xác định
UNIMPLEMENTED
Method chưa được implement
Trong gRPC, lỗi không trả qua response message, mà được đẩy qua channel error.
Ví dụ (Server - Java)
public void getUser(UserRequest request, StreamObserver<UserResponse> responseObserver) {
if (request.getUserId().isEmpty()) {
responseObserver.onError(
Status.INVALID_ARGUMENT
.withDescription("userId must not be empty")
.asRuntimeException()
);
return;
}
User user = findUser(request.getUserId());
if (user == null) {
responseObserver.onError(
Status.NOT_FOUND
.withDescription("User not found")
.asRuntimeException()
);
return;
}
responseObserver.onNext(toResponse(user));
responseObserver.onCompleted();
}
Client sẽ nhận lỗi thông qua exception:
try {
UserResponse res = blockingStub.getUser(request);
} catch (StatusRuntimeException e) {
Status.Code code = e.getStatus().getCode();
if (code == Status.Code.NOT_FOUND) {
// handle not found
} else if (code == Status.Code.INVALID_ARGUMENT) {
// handle validation error
}
}
gRPC Interceptor là cơ chế cho phép chèn logic vào trước hoặc sau khi một RPC được thực thi ở cả phía client và server.
Đặc điểm:
Chạy ở phía client trước khi request được gửi đi.
Client → ClientInterceptor → gRPC call → Server
public class AuthClientInterceptor implements ClientInterceptor {
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
MethodDescriptor<ReqT, RespT> method,
CallOptions callOptions,
Channel next) {
return new ForwardingClientCall.SimpleForwardingClientCall<>(
next.newCall(method, callOptions)) {
@Override
public void start(Listener<RespT> responseListener, Metadata headers) {
headers.put(
Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER),
"Bearer token-value"
);
super.start(responseListener, headers);
}
};
}
}
Chạy ở phía server trước khi request được xử lý bởi service.
Client → ServerInterceptor → Service Handler
public class AuthServerInterceptor implements ServerInterceptor {
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
ServerCall<ReqT, RespT> call,
Metadata headers,
ServerCallHandler<ReqT, RespT> next) {
String token = headers.get(
Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER)
);
if (token == null) {
call.close(Status.UNAUTHENTICATED, headers);
return new ServerCall.Listener<ReqT>() {};
}
return next.startCall(call, headers);
}
}
gRPC Metadata là cơ chế truyền dữ liệu dạng key–value kèm theo request/response, tương tự như HTTP headers.
Đặc điểm:
Các loại dữ liệu thường dùng:
authorization (JWT, token)x-user-idtrace-idrequest-idMetadata dùng để xử lý các thông tin ngoài business data:
trace-id xuyên suốt hệ thốngTrong gRPC, metadata được chia thành:
Client → Metadata (headers) → Server
Server → Metadata (headers/trailers) → Client
Metadata.Key<String> AUTHORIZATION =
Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
ASCII_STRING_MARSHALLERBINARY_BYTE_MARSHALLER (cho dữ liệu nhị phân)Metadata metadata = new Metadata();
metadata.put(AUTHORIZATION, "Bearer token-value");
stub = MetadataUtils.attachHeaders(stub, metadata);
String token = headers.get(AUTHORIZATION);
Metadata responseHeaders = new Metadata();
responseHeaders.put(
Metadata.Key.of("custom-header", Metadata.ASCII_STRING_MARSHALLER),
"value"
);
call.sendHeaders(responseHeaders);
bin nếu là binaryMetadata thường được xử lý thông qua interceptor:
Client → Interceptor → attach metadata → gửi request
Nhận request → Interceptor → đọc metadata → validate → handler
Một file .proto thường gồm các thành phần chính:
syntax = "proto3";
package user;
Ví dụ Java:
option java_package = "com.example.user";
option java_multiple_files = true;
message UserRequest {
string user_id = 1;
}
message UserResponse {
string name = 1;
int32 age = 2;
}
string, int32, …)service UserService {
rpc GetUser (UserRequest) returns (UserResponse);
}
Các kiểu dữ liệu thường dùng
Scalar types:
string, int32, int64, bool, bytes, …Repeated (array):
repeated string tags = 3;
Map:
map<string, string> metadata = 4;
Enum
enum Status {
UNKNOWN = 0;
ACTIVE = 1;
INACTIVE = 2;
}
message Order {
string id = 1;
message Item {
string name = 1;
int32 quantity = 2;
}
repeated Item items = 2;
}
.protosyntax = "proto3";
package example;
option java_package = "com.example";
message Request {
string id = 1;
}
message Response {
string result = 1;
}
service ExampleService {
rpc ExampleMethod (Request) returns (Response);
}