java通过gRPC整合tensorflow serving——gRPC java入门例子

Contents

1 java通过gRPC整合tensorflow serving——gRPC java入门例子
2 相关

java通过gRPC整合tensorflow serving——gRPC java入门例子

项目中以前需要把算法同事们train好的tensorflow model包装成服务提供给其他部门应用，一开始我们使用python直接调用tensorflow，然后java和python之间的交互通过jni或者rabbitmq的方式，这种方式的问题就是调用tensorflow的效率较低，不得不频繁的load model的checkpoint文件，推断执行慢，后来随着tensorflow官方推出tensorflow serving，我们也尝试了这种方式，发现确实可以很大程度上提升推断的性能，从一开始调用tensorflow每一次推断差不多要几秒钟（导致我们的服务一开始基本都是只能提供异步批量处理的方式），后来的tfserving的方式耗时降到了百毫秒的级别，使得同步的调用成为现实。

这几篇文章就简单讲一下在实践中，我们是如何通过gRPC整合tensorflow serving调用深度学习模型的。

打算分为以下几个部分

首先必须要知道一些gRPC的基本知识，能用java实现一个客户端和服务器端的小demo 就是本篇gRPC java入门例子
然后是 tensorflow model导出到tensorflow serving的要点讲解
接着是一个完整的java通过gRPC调用tensorflow serving的例子。

下面开始本篇，首先关于gRPC的介绍就不多说了，网上随便一搜都是，什么高性能，以支持移动和HTTP2.0为主的开源RPC协议什么的。。。

我们直接开始java的小demo，我基于Intellij的iDE讲解，eclipse大体也差不多。

1. 首先第一步，新建一个基于gradle构建的java项目，如图

java整合tensorflow serving

然后点下一步，填写maven坐标什么的，按照自己实际情况填写就好了，最后点finish，一个gradle项目就建好了。

2.接着加入相关的依赖和插件

因为我们需要gRPC的支持所以需要加入grpc几个相关的依赖，还有一个protobuf的gradle插件，用于将proto的协议文件生成java源码。

编辑build.gradle 如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

group 'com.xyz'
version '1.0-SNAPSHOT'

repositories {
mavenCentral()
}

buildscript {
repositories {
mavenCentral()
}
dependencies {
classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.3'
}
}

apply plugin: 'java'

apply plugin: 'com.google.protobuf'

sourceCompatibility = 1.8

protobuf {
generatedFilesBaseDir = "$projectDir/src/"
sourceSets {
main {
proto {
// 除了默认的'src/main/proto'目录新增proto文件的方法
srcDir 'src/main/protobuf'
include '**/*.protodevel'
}
}
}
protoc {
// The artifact spec for the Protobuf Compiler
artifact = 'com.google.protobuf:protoc:3.0.0'
}
plugins {
grpc {
artifact = 'io.grpc:protoc-gen-grpc-java:1.0.0-pre2'
}
}
generateProtoTasks {
all()*.plugins {
grpc {}
}
}
}

dependencies {
compile 'com.google.protobuf:protobuf-java:3.0.0'
compile 'io.grpc:grpc-stub:1.0.0-pre2'
compile 'io.grpc:grpc-protobuf:1.0.0-pre2'
compile "io.grpc:grpc-netty:1.0.0-pre2"
if (JavaVersion.current().isJava9Compatible()) {
compile 'javax.annotation:javax.annotation-api:1.3.1'
}
testCompile 'junit:junit:4.12'
}

3.接着就是编写我们的服务协议定义的proto文件

在src/main/目录下新建一个proto目录，默认gradle-protobuf插件会自动处理这个目录下的proto文件，如果你想自己定义其他目录，就要像上面那个配置文件中那样，添加proto–srcDir的配置单元了，这里我起一个名字叫test_grpc.proto 内容如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

// Copyright 2015, gRPC Authors
// All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
syntax = "proto3";

option java_multiple_files = true;
option java_package = "io.grpc.examples";
option java_outer_classname = "TestGRPCProto";

package examples;

service TestService {
rpc ListPeople (Query) returns (People) {//无办法不传参数只能定义一个空消息传进去 ,我想这可能是也是一种防御性编程吧毕竟后面可能会用到参数

}

}

message Query {

}

message People {
string name = 1;
int32 age = 2;
}

就是简单的定义了一个叫TestService的服务，然后只有一个方法叫ListPeople，方法接收一个空的参数，但是gRPC不允许这种完全空的，只能通过自己定义个空消息实现，ListPeople返回一个People的对象，都是写死的，做个演示。

保存后，执行gradle build，或者其他触发编译的命令，protoc就会将刚才的proto文件生成对应的java代码，拷贝这些代码到我们的src/main/java目录，下面就可以开始编写服务代码了。

4.然后我们编写服务端

新建一个叫TestServer.java的源文件，主要的步骤就是首先写一个服务类继承刚才gRPC自动生成的那堆代码中的服务stub类，就是XXXImplBase类，其中的XXX就是你proto中定义的服务名，覆写里面的服务方法后，然后再写一个启动server的代码就ok了，具体如下图，作为demo可以直接拷贝

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

package io.grpc.examples;

import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;

import java.io.IOException;

public class TestServer {

private Server server;

private void start() throws IOException {
/* The port on which the server should run */
int port = 50051;
server = ServerBuilder.forPort(port)
.addService(new TestServiceImpl())
.build()
.start();
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
System.err.println("*** shutting down gRPC server since JVM is shutting down");
TestServer.this.stop();
System.err.println("*** server shut down");
}
});
}

private void stop() {
if (server != null) {
server.shutdown();
}
}

/**
* Await termination on the main thread since the grpc library uses daemon threads.
*/
private void blockUntilShutdown() throws InterruptedException {
if (server != null) {
server.awaitTermination();
}
}

/**
* Main launches the server from the command line.
*/
public static void main(String[] args) throws IOException, InterruptedException {
final TestServer server = new TestServer();
server.start();
server.blockUntilShutdown();
}

static class TestServiceImpl extends TestServiceGrpc.TestServiceImplBase {
@Override
public void listPeople(Query request, StreamObserver responseObserver) {
People people = People.newBuilder().setAge(11).setName("xiaozhang").build();
responseObserver.onNext(people);
responseObserver.onCompleted();
}
}
}

5.最后就是client端的调用代码了

新建一个类TestClient.java 拷贝以下代码，其中调用gRPC的服务client分为两种，一种是blocking的同步调用，一种是non-blocking的异步调用，代码的注释中有这块，可以参考。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

package io.grpc.examples;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;

public class TestClient {
public static void main(String[] args) throws InterruptedException {
ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1",50051).usePlaintext(true).build();
TestServiceGrpc.TestServiceBlockingStub blockingStub = TestServiceGrpc.newBlockingStub(channel);
TestServiceGrpc.TestServiceStub ayncStub = TestServiceGrpc.newStub(channel);

//同步调用
System.out.println(blockingStub.listPeople(Query.newBuilder().build()));
//异步调用
ayncStub.listPeople(Query.newBuilder().build(), new StreamObserver(){
@Override
public void onNext(People value) {
System.out.println(value);
}

@Override
public void onError(Throwable t) {

}

@Override
public void onCompleted() {
System.out.println("finished");
}
});
System.out.println("异步调用不阻塞,所以这句先打印");
Thread.sleep(10);
}
}

最后启动server，再启动服务端就可以看到简单的效果了，时间仓促，有问题敬请留言。有了对java使用grpc的简单印象后，后面我们继续探讨整合tensorflow serving的方式。

java通过gRPC整合tensorflow serving——gRPC java入门例子