gRPC with Java: A Step-by-Step Introduction
Table of Contents
Configuring Maven
To get started with gRPC in Java, we first need to configure Maven to manage the dependencies for both the Protoc compiler and the protoc-gen-grpc-java plugin. This simplifies the process because Maven automatically downloads the necessary dependencies, without us having to manually install them. The Protoc compiler compiles the .proto files, while theprotoc-gen-grpc-java plugin generates the corresponding Java stubs.
Additionally, we use the kr.motd.maven:os-maven-plugin extension in our build configuration. This extension detects the operating system (OS) type and architecture, helping us select the appropriate version of the Protoc compiler based on the platform you're using (e.g., Windows, Linux, or macOS). The os.detected.classifier placeholder dynamically resolves to the OS-specific version, such as win64,linux, etc. This ensures the correct versions of the Protoc compiler and protoc-gen-grpc-java are used in your environment.
1 <dependencies>
2 <dependency>
3 <groupId>io.grpc</groupId>
4 <artifactId>grpc-netty</artifactId>
5 <version>1.62.2</version>
6 </dependency>
7 <dependency>
8 <groupId>io.grpc</groupId>
9 <artifactId>grpc-protobuf</artifactId>
10 <version>1.62.2</version>
11 </dependency>
12 <dependency>
13 <groupId>io.grpc</groupId>
14 <artifactId>grpc-stub</artifactId>
15 <version>1.62.2</version>
16 </dependency>
17 <dependency>
18 <groupId>io.grpc</groupId>
19 <artifactId>protoc-gen-grpc-java</artifactId>
20 <version>1.60.1</version>
21 <type>pom</type>
22 </dependency>
23 <dependency>
24 <groupId>javax.annotation</groupId>
25 <artifactId>javax.annotation-api</artifactId>
26 <version>1.3.2</version>
27 </dependency>
28 </dependencies>
29
30 <build>
31 <extensions>
32 <extension>
33 <groupId>kr.motd.maven</groupId>
34 <artifactId>os-maven-plugin</artifactId>
35 <version>1.6.2</version>
36 </extension>
37 </extensions>
38 <plugins>
39 <plugin>
40 <groupId>org.xolstice.maven.plugins</groupId>
41 <artifactId>protobuf-maven-plugin</artifactId>
42 <version>0.6.1</version>
43 <configuration>
44 <protocArtifact>
45 com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier}
46 </protocArtifact>
47 <pluginId>grpc-java</pluginId>
48 <pluginArtifact>
49 io.grpc:protoc-gen-grpc-java:1.4.0:exe:${os.detected.classifier}
50 </pluginArtifact>
51 <protoSourceRoot>src/main/java/service</protoSourceRoot>
52 </configuration>
53 <executions>
54 <execution>
55 <goals>
56 <goal>compile</goal>
57 <goal>compile-custom</goal>
58 </goals>
59 </execution>
60 </executions>
61 </plugin>
62 </plugins>
63 </build>Writing the proto service
In this example, we will create a gRPC task server that tracks the frequency of characters it receives. The server will handle aTaskRequest containing a word and its count, and in return, it will respond with a message. Additionally, you can send a TaskStatusRequestto check the progress of a task. The server will reply with aTaskStatusResponse, which includes an enum that indicates whether the task is pending, in progress, completed, or failed. Finally, we will implement agetTaskResult method, which returns a map of words and their corresponding frequencies.
1syntax = "proto3";
2option java_multiple_files = true;
3package service;
4
5import "google/protobuf/empty.proto";
6import "google/protobuf/wrappers.proto";
7
8message TaskRequest {
9 string id = 1;
10 string word = 2;
11 google.protobuf.Int32Value count = 3;
12}
13
14message TaskResponse {
15 string response = 1;
16}
17
18message TaskStatusRequest {
19 string id = 1;
20}
21
22message TaskStatusResponse {
23 enum Status {
24 PENDING = 0;
25 IN_PROGRESS = 1;
26 COMPLETED = 2;
27 FAILED = 3;
28 }
29 Status status = 1;
30}
31
32message Result {
33 map<string, int32> pairs = 1;
34}
35
36service TaskService {
37 rpc submitTask(TaskRequest) returns (TaskResponse);
38 rpc getTaskStatus(TaskStatusRequest) returns (TaskStatusResponse);
39 rpc getTaskResult(google.protobuf.Empty) returns (Result);
40}Once the .proto file is defined, we can run mvn clean install to compile the .proto file and automatically generate the necessary Java stubs.
Implementing our Service
After compiling the .proto files and generating the Java classes, the next step is to implement the service. In this implementation, we need to override the submitTask method, which takes a TaskRequest object as input, along with a StreamObserver<TaskResponse>, which is used to send the response back to the client.
Within the submitTask method, we use anExecutorService to handle the task asynchronously. This involves updating two concurrent data structures—countMap and taskStatus—which track the word count and task status, respectively. The task is then executed in a separate thread, and we simulate a delay to represent long-running tasks.
1public class TaskServiceImpl extends TaskServiceGrpc.TaskServiceImplBase implements AutoCloseable {
2
3 ConcurrentHashMap<String, Integer> countMap = new ConcurrentHashMap<>();
4 ConcurrentHashMap<String, TaskStatusResponse.Status> taskStatus = new ConcurrentHashMap<>();
5 ExecutorService executor = Executors.newFixedThreadPool(10);
6 private static final int LONG_TASK_DELAY = 5;
7
8 @Override
9 public void submitTask(
10 TaskRequest request, StreamObserver<TaskResponse> responseObserver) {
11
12 String taskId = request.getId();
13 String word = request.getWord();
14 Int32Value count = request.getCount();
15 Runnable runnableTask = () -> {
16 try {
17 countMap.compute(word, (k, v) -> (v == null ? 0 : v) + count.getValue());
18 taskStatus.compute(taskId, (key, oldValue) -> TaskStatusResponse.Status.IN_PROGRESS);
19 TimeUnit.SECONDS.sleep(LONG_TASK_DELAY);
20 taskStatus.compute(taskId, (key, oldValue) -> TaskStatusResponse.Status.COMPLETED);
21 } catch (InterruptedException e) {
22 Thread.currentThread().interrupt(); // set the interrupt flag
23 taskStatus.put(taskId, TaskStatusResponse.Status.FAILED);
24 e.printStackTrace();
25 }
26 };
27 executor.submit(runnableTask);
28 String result = "Task is successfully submitted to thread pool";
29 TaskResponse response = TaskResponse.newBuilder()
30 .setResponse(result)
31 .build();
32 responseObserver.onNext(response);
33 responseObserver.onCompleted();
34 }
35}Running the grpc server
In this section, we'll set up and run a gRPC server in Java. The server is built using a Server object, which is configured to listen on a specific port and host the required services. To ensure proper resource cleanup during shutdown, we also add a shutdown hook.
1public class GrpcServer {
2 public static void main(String[] args) throws IOException, InterruptedException {
3 Server server = ServerBuilder
4 .forPort(8080)
5 .addService(new TaskServiceImpl())
6 .build();
7
8 server.start();
9 Runtime.getRuntime().addShutdownHook(new Thread(() -> { // shut down hook, ensures that when the JVM is shutting down, it will run the specified cleanup logic before exiting
10 System.out.println("Shutting down gRPC server...");
11 server.shutdown();
12 System.out.println("Server shut down.");
13 }));
14
15 server.awaitTermination();
16 }
17}Creating gRPC Client Stubs
In this section, we'll create a gRPC client to interact with the server. We use a newBlockingStub to simulate synchronous communication. The client sends multiple TaskRequest messages to the server and waits for responses. After a 10-second delay, it retrieves the final results using the getTaskResult method.
1public class GrpcClient {
2 public static void main(String[] args) throws InterruptedException {
3 ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 8080)
4 .usePlaintext()
5 .build();
6 TaskServiceGrpc.TaskServiceBlockingStub taskStub = TaskServiceGrpc.newBlockingStub(channel);
7 for(int i = 0; i < 10; i++) {
8 char c = (char) ((Math.random() * 5) + 'a');
9 String word = String.valueOf(c);
10 int count = (int) (Math.random() * 10);
11 TaskResponse taskResponse = taskStub.submitTask(TaskRequest.newBuilder()
12 .setCount(Int32Value.of(count))
13 .setId(String.valueOf(i))
14 .setWord(word).build());
15 System.out.println("Response : " + taskResponse.getResponse());
16 }
17 TimeUnit.SECONDS.sleep(10);
18 Result res = taskStub.getTaskResult(Empty.newBuilder().build());
19 System.out.println(res);
20 channel.shutdown();
21 }
22}