gRPC

gRPC

gRPCは、Googleが開発した高性能なRPC（Remote Procedure Call）フレームワークです。Spring Bootでは、grpc-spring-boot-starterを使用してgRPCを実装できます。

なぜgRPCが必要なのか

REST APIとの比較

REST APIの課題:

// REST API: JSON over HTTP
POST /api/users
Content-Type: application/json

{
  "name": "John Doe",
  "email": "john@example.com"
}

// 問題点:
// - テキストベース（JSON）のためサイズが大きい
// - HTTP/1.1のため、複数のリクエストを並列処理できない
// - 型安全性が低い（実行時エラー）

gRPCの解決:

// Protocol Buffers: バイナリ形式、型安全
service UserService {
  rpc CreateUser(CreateUserRequest) returns (User);
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
}

// メリット:
// - バイナリ形式のためサイズが小さい
// - HTTP/2のため、複数のリクエストを並列処理できる
// - 型安全性が高い（コンパイル時エラー）

メリット:

1. 高性能: バイナリ形式とHTTP/2による高速通信
2. 型安全性: Protocol Buffersによる型安全なAPI
3. ストリーミング: 双方向ストリーミング対応
4. 言語非依存: 複数の言語で実装可能

Protocol Buffers

Protocol Buffers（protobuf）は、構造化データをシリアライズするためのフォーマットです。

.protoファイルの定義

src/main/proto/user.proto:

syntax = "proto3";

package com.example.grpc;

option java_package = "com.example.grpc";
option java_outer_classname = "UserProto";

// ユーザーサービス
service UserService {
  // 単一リクエスト/レスポンス
  rpc GetUser(GetUserRequest) returns (User);

  // サーバーストリーミング
  rpc ListUsers(ListUsersRequest) returns (stream User);

  // クライアントストリーミング
  rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);

  // 双方向ストリーミング
  rpc ChatUsers(stream ChatMessage) returns (stream ChatMessage);
}

// リクエストメッセージ
message GetUserRequest {
  int64 id = 1;
}

message ListUsersRequest {
  int32 page = 1;
  int32 size = 2;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
}

message CreateUsersResponse {
  int32 count = 1;
  repeated User users = 2;
}

message ChatMessage {
  int64 user_id = 1;
  string message = 2;
}

// レスポンスメッセージ
message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
  int64 created_at = 4;
}

Spring Boot gRPCの設定

依存関係の追加

Maven (pom.xml):

<dependencies>
    <dependency>
        <groupId>net.devh</groupId>
        <artifactId>grpc-spring-boot-starter</artifactId>
        <version>2.15.0.RELEASE</version>
    </dependency>
</dependencies>

<build>
    <extensions>
        <extension>
            <groupId>kr.motd.maven</groupId>
            <artifactId>os-maven-plugin</artifactId>
            <version>1.7.0</version>
        </extension>
    </extensions>
    <plugins>
        <plugin>
            <groupId>org.xolstice.maven.plugins</groupId>
            <artifactId>protobuf-maven-plugin</artifactId>
            <version>0.6.1</version>
            <configuration>
                <protocArtifact>com.google.protobuf:protoc:3.21.1:exe:${os.detected.classifier}</protocArtifact>
                <pluginId>grpc-java</pluginId>
                <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.50.0:exe:${os.detected.classifier}</pluginArtifact>
            </configuration>
            <executions>
                <execution>
                    <goals>
                        <goal>compile</goal>
                        <goal>compile-custom</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

Gradle (build.gradle):

plugins {
    id 'com.google.protobuf' version '0.9.2'
}

dependencies {
    implementation 'net.devh:grpc-spring-boot-starter:2.15.0.RELEASE'
}

protobuf {
    protoc {
        artifact = "com.google.protobuf:protoc:3.21.1"
    }
    plugins {
        grpc {
            artifact = "io.grpc:protoc-gen-grpc-java:1.50.0"
        }
    }
    generateProtoTasks {
        all()*.plugins {
            grpc {}
        }
    }
}

基本設定

application.yml:

grpc:
  server:
    port: 9090

gRPCサービスの実装

単一リクエスト/レスポンス

@GrpcService
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {

    private final UserRepository userRepository;

    public UserServiceImpl(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    @Override
    public void getUser(GetUserRequest request,
                       StreamObserver<User> responseObserver) {
        User user = userRepository.findById(request.getId())
            .map(this::toProtoUser)
            .orElseThrow(() -> Status.NOT_FOUND
                .withDescription("User not found")
                .asRuntimeException());

        responseObserver.onNext(user);
        responseObserver.onCompleted();
    }

    private User toProtoUser(com.example.entity.User entity) {
        return User.newBuilder()
            .setId(entity.getId())
            .setName(entity.getName())
            .setEmail(entity.getEmail())
            .setCreatedAt(entity.getCreatedAt().toEpochMilli())
            .build();
    }
}

サーバーストリーミング

@GrpcService
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {

    @Override
    public void listUsers(ListUsersRequest request,
                         StreamObserver<User> responseObserver) {
        int page = request.getPage();
        int size = request.getSize();

        List<com.example.entity.User> users = userRepository
            .findAll(PageRequest.of(page, size))
            .getContent();

        for (com.example.entity.User user : users) {
            responseObserver.onNext(toProtoUser(user));
        }

        responseObserver.onCompleted();
    }
}

クライアントストリーミング

@GrpcService
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {

    @Override
    public StreamObserver<CreateUserRequest> createUsers(
            StreamObserver<CreateUsersResponse> responseObserver) {

        return new StreamObserver<CreateUserRequest>() {
            private final List<User> createdUsers = new ArrayList<>();

            @Override
            public void onNext(CreateUserRequest request) {
                com.example.entity.User user = new com.example.entity.User();
                user.setName(request.getName());
                user.setEmail(request.getEmail());
                user = userRepository.save(user);
                createdUsers.add(toProtoUser(user));
            }

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

            @Override
            public void onCompleted() {
                CreateUsersResponse response = CreateUsersResponse.newBuilder()
                    .setCount(createdUsers.size())
                    .addAllUsers(createdUsers)
                    .build();
                responseObserver.onNext(response);
                responseObserver.onCompleted();
            }
        };
    }
}

双方向ストリーミング

@GrpcService
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {

    @Override
    public StreamObserver<ChatMessage> chatUsers(
            StreamObserver<ChatMessage> responseObserver) {

        return new StreamObserver<ChatMessage>() {
            @Override
            public void onNext(ChatMessage message) {
                // メッセージを受信して処理
                ChatMessage reply = ChatMessage.newBuilder()
                    .setUserId(message.getUserId())
                    .setMessage("Echo: " + message.getMessage())
                    .build();
                responseObserver.onNext(reply);
            }

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

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

gRPCクライアントの実装

@Service
public class UserClientService {

    private final UserServiceGrpc.UserServiceBlockingStub userServiceStub;

    public UserClientService(@GrpcClient("user-service")
                            Channel channel) {
        this.userServiceStub = UserServiceGrpc.newBlockingStub(channel);
    }

    public User getUser(Long id) {
        GetUserRequest request = GetUserRequest.newBuilder()
            .setId(id)
            .build();

        return userServiceStub.getUser(request);
    }

    public List<User> listUsers(int page, int size) {
        ListUsersRequest request = ListUsersRequest.newBuilder()
            .setPage(page)
            .setSize(size)
            .build();

        Iterator<User> users = userServiceStub.listUsers(request);
        List<User> result = new ArrayList<>();
        while (users.hasNext()) {
            result.add(users.next());
        }
        return result;
    }
}

エラーハンドリング

@GrpcService
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {

    @Override
    public void getUser(GetUserRequest request,
                       StreamObserver<User> responseObserver) {
        try {
            User user = userRepository.findById(request.getId())
                .map(this::toProtoUser)
                .orElseThrow(() -> new UserNotFoundException());

            responseObserver.onNext(user);
            responseObserver.onCompleted();

        } catch (UserNotFoundException e) {
            responseObserver.onError(Status.NOT_FOUND
                .withDescription("User not found: " + request.getId())
                .asRuntimeException());
        } catch (Exception e) {
            responseObserver.onError(Status.INTERNAL
                .withDescription("Internal server error")
                .withCause(e)
                .asRuntimeException());
        }
    }
}

インターセプター（認証・認可）

@GrpcGlobalInterceptor
public class AuthInterceptor 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 || !isValidToken(token)) {
            call.close(Status.UNAUTHENTICATED
                .withDescription("Invalid token"),
                new Metadata());
            return new ServerCall.Listener<ReqT>() {};
        }

        // 認証情報をコンテキストに設定
        Context context = Context.current()
            .withValue(AUTH_CONTEXT_KEY, getAuthInfo(token));

        return Contexts.interceptCall(context, call, headers, next);
    }

    private boolean isValidToken(String token) {
        // JWT検証などの実装
        return true;
    }
}

まとめ

gRPCを使用したRPC通信のポイント：

• 高性能: バイナリ形式とHTTP/2による高速通信
• 型安全性: Protocol Buffersによる型安全なAPI
• ストリーミング: 単方向・双方向ストリーミング対応
• エラーハンドリング: Statusによる適切なエラー処理
• インターセプター: 認証・認可などの横断的関心事

gRPCは、マイクロサービス間の通信や高性能が求められるAPIで非常に有用です。適切に実装することで、アプリケーションのパフォーマンスと型安全性を大幅に向上させることができます。