gRPC Example
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| syntax = "proto3";
option java_multiple_files = true;
option java_package = "io.grpc.hello";
option java_outer_classname = "HelloProto";
option objc_class_prefix = "HLW";
package hello;
service HelloService{
rpc Say(HelloRequest) returns (HelloReply) {}
}
message HelloRequest {
string name = 1;
}
message HelloReply {
string message = 1;
}
|
Client.java
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
| package io.grpc.hello;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.StatusRuntimeException;
import java.util.concurrent.TimeUnit;
public class HelloWorldClient {
private final ManagedChannel channel;
private final HelloServiceGrpc.HelloServiceBlockingStub blockingStub;
/**
* construct network channel
**/
public HelloWorldClient(String host, int port) {
this(ManagedChannelBuilder.forAddress(host, port)
.usePlaintext()
.build());
}
/**
* build stub for sending request
**/
HelloWorldClient(ManagedChannel channel) {
this.channel = channel;
blockingStub = HelloServiceGrpc.newBlockingStub(channel);
}
/**
* resource destruction
**/
public void shutdown() throws InterruptedException {
channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
}
/**
* send rpc request
**/
public void say(String name) {
HelloRequest request = HelloRequest.newBuilder().setName(name).build();
HelloReply response;
try {
response = blockingStub.say(request);
} catch (StatusRuntimeException e) {
return;
}
System.out.println(response);
}
public static void main(String[] args) throws Exception {
HelloWorldClient client = new HelloWorldClient("127.0.0.1", 50051);
try {
client.say("world");
} finally {
client.shutdown();
}
}
}
|
Server.java
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
63
64
65
| package io.grpc.hello;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
public class HelloWorldServer {
static class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase {
@Override
public void say(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
responseObserver.onNext(reply);
responseObserver.onCompleted();
}
}
private Server server;
/**
* expose service
**/
private void start() throws IOException {
int port = 50051;
server = ServerBuilder.forPort(port)
.addService(new HelloServiceImpl())
.build()
.start();
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
HelloWorldServer.this.stop();
}
});
}
/**
* resource destruction
**/
private void stop() {
if (server != null) {
server.shutdown();
}
}
/**
* resource destruction
**/
private void blockUntilShutdown() throws InterruptedException {
if (server != null) {
server.awaitTermination();
}
}
public static void main(String[] args) throws IOException, InterruptedException {
final HelloWorldServer server = new HelloWorldServer();
server.start();
server.blockUntilShutdown();
}
}
|
Dynamic proxy
RPC will automatically generate a proxy class for the interface.
When we inject the interface into the project, the actual binding process is the proxy class generated by the interface. This way, when the interface method is called, it is intercepted by the generated proxy class so that we can add remote call logic to the generated proxy class. By this technique, we can help users to shield the details of the remote call, and achieve the experience of calling the remote as if it were a local call.
Let’s take a look at the following example:
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
| public interface Hello {
String say();
}
public class RealHello {
public String invoke(){
return "I'm proxy";
}
}
// Autogenerated MyProxy class hides the details of remote call into RealHello
public class MyProxy implements InvocationHandler {
private Object target;
MyProxy(Object target) {
this.target = target;
}
@Override
public Object invoke(Object proxy, Method method, Object[] paramValues) {
return ((RealHello)target).invoke();
}
}
public class TestProxy {
public static void main(String[] args){
MyProxy proxy = new MyProxy(new RealHello());
ClassLoader classLoader = ClassLoaderUtils.getCurrentClassLoader();
// this line binds the Hello interface to the MyProxy
Hello test = (Hello) Proxy.newProxyInstance(classLoader, new Class[]{Hello.class}, proxy);
System.out.println(test.say()); // this prints "I'm proxy"
}
}
|
Implementing a framework
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| public class RpcRequest {
private String requestId;
private String className;
private String methodName;
private Class<?>[] parameterTypes;
private Object[] parameters;
// getter/setter...
}
public class RpcResponse {
private String requestId;
private Throwable error;
private Object result;
// getter/setter...
}
|
Serializaion (use Hessian)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| import com.caucho.hessian.io.HessianInput;
import com.caucho.hessian.io.HessianOutput;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
public class Serializer {
public static <T> byte[] serialize(T input) throws IOException {
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
HessianOutput hessianOutput = new HessianOutput(byteArrayOutputStream);
hessianOutput.writeObject(input);
return byteArrayOutputStream.toByteArray();
}
public static <T> T deserialize(byte[] input) throws IOException {
ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(input);
HessianInput hessianInput = new HessianInput(byteArrayInputStream);
return (T) hessianInput.readObject();
}
}
|
Service discovery
Request handle
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
63
64
65
66
| public class RequestHandler extends SimpleChannelInboundHandler<RpcRequest> {
private static final Logger LOGGER = LoggerFactory.getLogger(RpcHandler.class);
private Map<String, Object> serviceMap;
public RequestHandler(Map<String, Object> serviceMap) {
this.serviceMap = serviceMap;
}
@Override
public void channelRead0(final ChannelHandlerContext ctx, RpcRequest request) throws Exception {
RpcResponse response = new RpcResponse();
response.setRequestId(request.getRequestId());
try {
Object result = handle(request);
response.setResult(result);
} catch (Throwable t) {
response.setError(t);
}
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
private RpcResponse handle(RpcRequest request) {
RpcResponse response = new RpcResponse();
response.setId(request.getId());
Object instance = serviceMap.get(request.getInterfaceName());
if (instance == null) {
return null;
}
try {
Class<?>[] types = Arrays.stream(request.getArguments())
.map(Argument::getType)
.toArray(Class<?>[]::new);
Method method = instance.getClass()
.getMethod(request.getMethodName(), types);
Object[] args = Arrays.stream(request.getArguments())
.map(Argument::getValue)
.toArray(Object[]::new);
Object result = method.invoke(instance, args);
response.setResponse(result);
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
return null;
}
return response;
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws Exception {
if (event instanceof IdleStateEvent) {
IdleStateEvent e = (IdleStateEvent) event;
if (e.state() == IdleState.READER_IDLE) {
ctx.channel().close();
}
} else {
super.userEventTriggered(ctx, event);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
LOGGER.error("server caught exception", cause);
ctx.close();
}
...
}
|
RPC service
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
63
64
65
| public class RpcServer implements ApplicationContextAware, InitializingBean {
private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);
private String serverAddress;
private ServiceRegistry serviceRegistry;
private Map<String, Object> handlerMap = new HashMap<>();
public RpcServer(String serverAddress) {
this.serverAddress = serverAddress;
}
public RpcServer(String serverAddress, ServiceRegistry serviceRegistry) {
this.serverAddress = serverAddress;
this.serviceRegistry = serviceRegistry;
}
@Override
public void setApplicationContext(ApplicationContext ctx) throws BeansException {
if (MapUtils.isNotEmpty(serviceBeanMap)) {
for (Object serviceBean : serviceBeanMap.values()) {
String interfaceName = serviceBean.getClass().getAnnotation(RpcService.class).value().getName();
handlerMap.put(interfaceName, serviceBean);
}
}
}
@Override
public void afterPropertiesSet() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel channel) throws Exception {
channel.pipeline()
.addLast(new RpcDecoder(RpcRequest.class))
.addLast(new RpcEncoder(RpcResponse.class))
.addLast(new RequestHandler(handlerMap));
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
String[] array = serverAddress.split(":");
String host = array[0];
int port = Integer.parseInt(array[1]);
ChannelFuture future = bootstrap.bind(host, port).sync();
LOGGER.debug("server started on port {}", port);
if (serviceRegistry != null) {
serviceRegistry.register(serverAddress);
}
future.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
|
Load balancing
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| public class RoundRobinLoadBalancer implements LoadBalancer {
private AtomicInteger current = new AtomicInteger(0);
@Override
public RpcProtocol route(String serviceKey, Map<Connection, RpcClient> connectedServerNodes) throws Exception {
Map<String, List<Connection>> serviceMap = getServiceMap(connectedServerNodes);
List<Connection> addressList = serviceMap.get(serviceKey);
if (addressList != null && addressList.size() > 0) {
int size = addressList.size();
int index = (current.getAndAdd(1) + size) % size;
return addressList.get(index);
} else {
throw new Exception("Can not find connection for service: " + serviceKey);
}
}
}
|
Dynamic proxy
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
| public class RpcProxy {
private String serverAddress;
private ServiceDiscovery serviceDiscovery;
public RpcProxy(String serverAddress) {
this.serverAddress = serverAddress;
}
public RpcProxy(ServiceDiscovery serviceDiscovery) {
this.serviceDiscovery = serviceDiscovery;
}
public <T> T create(Class<?> interfaceClass) {
return (T) Proxy.newProxyInstance(
interfaceClass.getClassLoader(),
new Class<?>[]{interfaceClass},
new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
RpcRequest request = new RpcRequest();
request.setRequestId(UUID.randomUUID().toString());
request.setClassName(method.getDeclaringClass().getName());
request.setMethodName(method.getName());
request.setParameterTypes(method.getParameterTypes());
request.setParameters(args);
if (serviceDiscovery != null) {
serverAddress = serviceDiscovery.discover();
}
String[] array = serverAddress.split(":");
String host = array[0];
int port = Integer.parseInt(array[1]);
RpcClient client = new RpcClient(host, port);
RpcResponse response = client.send(request);
if (response.isError()) {
throw response.getError();
} else {
return response.getResult();
}
}
}
);
}
}
|
Client
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
63
| public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {
private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);
private String host;
private int port;
private RpcResponse response;
private final Object obj = new Object();
public RpcClient(String host, int port) {
this.host = host;
this.port = port;
}
@Override
public void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
this.response = response;
synchronized (obj) {
obj.notifyAll();
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
LOGGER.error("client caught exception", cause);
ctx.close();
}
public RpcResponse send(RpcRequest request) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel channel) throws Exception {
channel.pipeline()
.addLast(new RpcEncoder(RpcRequest.class))
.addLast(new RpcDecoder(RpcResponse.class))
.addLast(RpcClient.this);
}
})
.option(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture future = bootstrap.connect(host, port).sync();
future.channel().writeAndFlush(request).sync();
synchronized (obj) {
obj.wait();
}
if (response != null) {
future.channel().closeFuture().sync();
}
return response;
} finally {
group.shutdownGracefully();
}
}
}
|