前言

grpc发送过程总结起来可以分为两步，一步是写入内存，也就是写入ByteBuf，写入ByteBuf的目的是因为Java NIO对于数据有要求，第二步就是通过netty发送数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
private static <ReqT, RespT> void asyncUnaryRequestCall(
    ClientCall<ReqT, RespT> call,
    ReqT req,
    StartableListener<RespT> responseListener) {
    startCall(call, responseListener);
    try {
        call.sendMessage(req);
        call.halfClose();
    } catch (RuntimeException e) {
        throw cancelThrow(call, e);
    } catch (Error e) {
        throw cancelThrow(call, e);
    }
}

call.sendMessage(req);这句命令就是处理我们的数据，而在正式处理我们发送的数据之前，会调用startCall(call, responseListener);这个命令，这个命令其中一个功能就是写入http2头部，写入头部的逻辑最后会调用下面的方法

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
private void writeHeadersInternal(Metadata headers, byte[] requestPayload) {
    AsciiString defaultPath = (AsciiString)NettyClientStream.methodDescriptorAccessor.geRawMethodName(NettyClientStream.this.method);
    if (defaultPath == null) {
        defaultPath = new AsciiString("/" + NettyClientStream.this.method.getFullMethodName());
        NettyClientStream.methodDescriptorAccessor.setRawMethodName(NettyClientStream.this.method, defaultPath);
    }

    boolean get = requestPayload != null;
    AsciiString httpMethod;
    if (get) {
        defaultPath = new AsciiString(defaultPath + "?" + BaseEncoding.base64().encode(requestPayload));
        httpMethod = Utils.HTTP_GET_METHOD;
    } else {
        httpMethod = Utils.HTTP_METHOD;
    }

    Http2Headers http2Headers = Utils.convertClientHeaders(headers, NettyClientStream.this.scheme, defaultPath, NettyClientStream.this.authority, httpMethod, NettyClientStream.this.userAgent);
    ChannelFutureListener failureListener = new ChannelFutureListener() {
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                Status s = NettyClientStream.this.transportState().handler.getLifecycleManager().getShutdownStatus();
                if (s == null) {
                    s = NettyClientStream.this.transportState().statusFromFailedFuture(future);
                }

                if (NettyClientStream.this.transportState().isNonExistent()) {
                    NettyClientStream.this.transportState().transportReportStatus(s, RpcProgress.MISCARRIED, true, new Metadata());
                } else {
                    NettyClientStream.this.transportState().transportReportStatus(s, RpcProgress.PROCESSED, true, new Metadata());
                }
            }

        }
    };
    NettyClientStream.this.writeQueue.enqueue(new CreateStreamCommand(http2Headers, NettyClientStream.this.transportState(), NettyClientStream.this.shouldBeCountedForInUse(), get), !NettyClientStream.this.method.getType().clientSendsOneMessage() || get).addListener(failureListener);
}

该方法就是创建一个CreateStreamCommand，并放入队列中，消息体后续也会加入这个队列，需要注意，如果这里是多元流或者没有消息体，那么将会直接flsuh，即会直接发送，否则会等到消息体加入队列之后，或者主动调用flush方法。

写入内存

这一步总结一下就是数据写入ByteBuf的过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
final void sendMessage(final ReqT message) {
  State savedState = state;
  if (savedState.passThrough) {
    savedState.winningSubstream.stream.writeMessage(method.streamRequest(message));
    return;
  }

  class SendMessageEntry implements BufferEntry {
    @Override
    public void runWith(Substream substream) {
      substream.stream.writeMessage(method.streamRequest(message));
      // TODO(ejona): Workaround Netty memory leak. Message writes always need to be followed by
      // flushes (or half close), but retry appears to have a code path that the flushes may
      // not happen. The code needs to be fixed and this removed. See #9340.
      substream.stream.flush();
    }
  }

  delayOrExecute(new SendMessageEntry());
}

这里面的runwith方法，其中substream.stream.writeMessage(method.streamRequest(message));就是将数据写入内存，而substream.stream.flush();就可以理解为发送数据

1
2
3
4
5
6
public void writeTo(OutputStream output) throws IOException {
    int bufferSize = CodedOutputStream.computePreferredBufferSize(this.getSerializedSize());
    CodedOutputStream codedOutput = CodedOutputStream.newInstance(output, bufferSize);
    this.writeTo(codedOutput);
    codedOutput.flush();
}

这个方法是将消息转为protobuf，使用protobuf也是grpc效率高的一个原因，AbstractMessageLite也就是grpc生成的消息类的父类，所以在这里调用writeTo方法的其实就是grpc根据我们定义数据生成的类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private void writeRaw(byte[] b, int off, int len) {
    while (len > 0) {
        if (buffer != null && buffer.writableBytes() == 0) {
            commitToSink(false, false);
        }
        if (buffer == null) {
            // Request a buffer allocation using the message length as a hint.
            buffer = bufferAllocator.allocate(len);
        }
        int toWrite = min(len, buffer.writableBytes());
        buffer.write(b, off, toWrite);
        off += toWrite;
        len -= toWrite;
    }
}

可以看到写入内存实际上就是写入MessageFramer类的buffer变量中，后续发送时，就是发送buffer变量

1
2
3
4
5
6
public ByteBuf writeBytes(byte[] src, int srcIndex, int length) {
    this.ensureWritable(length);
    this.setBytes(this.writerIndex, src, srcIndex, length);
    this.writerIndex += length;
    return this;
}

后面是调用了netty ByteBuf的writeBytes方法，就是将数据写入到ByteBuf

1
public native void copyMemory(Object var1, long var2, Object var4, long var5, long var7);

调用到这里其实已经是netty的范围了，ByteBuf最后会调用到一个本地方法，copyMemory的五个参数分别是

  • srcBase：源对象的基地址，即要复制数据的起始位置。
  • srcOffset：源对象的偏移量，即要复制数据的起始偏移量。
  • destBase：目标对象的基地址，即要复制数据的目标位置。
  • destOffset：目标对象的偏移量，即要复制数据的目标偏移量。
  • length：要复制的数据长度。

发送数据

从上面io.grpc.internal.RetriableStream#sendMessage方法的runwith方法中的substream.stream.flush();，就是开启刷新步骤，也就是发送流程

1
2
3
4
5
6
private void commitToSink(boolean endOfStream, boolean flush) {
    WritableBuffer buf = buffer;
    buffer = null;
    sink.deliverFrame(buf, endOfStream, flush, messagesBuffered);
    messagesBuffered = 0;
}

这个方法将上面提到的的buffer写入sink，就是正式发送数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private void writeFrameInternal(WritableBuffer frame, boolean endOfStream, boolean flush, final int numMessages) {
    Preconditions.checkArgument(numMessages >= 0);
    ByteBuf bytebuf = frame == null ? Unpooled.EMPTY_BUFFER : ((NettyWritableBuffer)frame).bytebuf().touch();
    final int numBytes = bytebuf.readableBytes();
    if (numBytes > 0) {
        NettyClientStream.this.onSendingBytes(numBytes);
        NettyClientStream.this.writeQueue.enqueue(new SendGrpcFrameCommand(NettyClientStream.this.transportState(), bytebuf, endOfStream), flush).addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess() && NettyClientStream.this.transportState().http2Stream() != null) {
                    NettyClientStream.this.transportState().onSentBytes(numBytes);
                    NettyClientStream.this.getTransportTracer().reportMessageSent(numMessages);
                }

            }
        });
    } else {
        NettyClientStream.this.writeQueue.enqueue(new SendGrpcFrameCommand(NettyClientStream.this.transportState(), bytebuf, endOfStream), flush);
    }

}

这里将消息包装成SendGrpcFrameCommand放入队列中进行处理，这里的队列就是最上面提到过http2header也加入的队列

1
2
3
4
5
6
7
8
9
10
11
ChannelFuture enqueue(QueuedCommand command, boolean flush) {
    Preconditions.checkArgument(command.promise() == null, "promise must not be set on command");
    ChannelPromise promise = this.channel.newPromise();
    command.promise(promise);
    this.queue.add(command);
    if (flush) {
        this.scheduleFlush();
    }

    return promise;
}

这个方法就是将消息放入队列并处理的方法，在上面http2header添加的时候，flush要根据条件判断，但是在添加我们的消息的时候，flush就是true，直接调用刷新方法，而this.scheduleFlush();命令会从队列中取出封装好的消息体，并调用他们的run方法

1
2
3
public final void run(Channel channel) {
channel.write(this, this.promise);
}
1
2
3
public final void run(Channel channel) {
channel.write(this, this.promise);
}

无论是http2header的CreateStreamCommand，还是我们写入消息的SendGrpcFrameCommand，他们的run方法都是一样的，都是调用netty的channel的write方法进行发送，至于channel的flush方法实在前面提到的scheduleFlush方法中执行的。我们知道channel需要添加channelHandler才能处理消息，而grpc实现了一个NettyClientHandler，下面就是NettyClientHandler的写入方法的实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (msg instanceof CreateStreamCommand) {
        this.createStream((CreateStreamCommand)msg, promise);
    } else if (msg instanceof SendGrpcFrameCommand) {
        this.sendGrpcFrame(ctx, (SendGrpcFrameCommand)msg, promise);
    } else if (msg instanceof CancelClientStreamCommand) {
        this.cancelStream(ctx, (CancelClientStreamCommand)msg, promise);
    } else if (msg instanceof SendPingCommand) {
        this.sendPingFrame(ctx, (SendPingCommand)msg, promise);
    } else if (msg instanceof GracefulCloseCommand) {
        this.gracefulClose(ctx, (GracefulCloseCommand)msg, promise);
    } else if (msg instanceof ForcefulCloseCommand) {
        this.forcefulClose(ctx, (ForcefulCloseCommand)msg, promise);
    } else {
        if (msg != NOOP_MESSAGE) {
            throw new AssertionError("Write called for unexpected type: " + msg.getClass().getName());
        }

        ctx.write(Unpooled.EMPTY_BUFFER, promise);
    }

}

总结

该篇讲到了我们自己写入的消息是如何发送出去的，但是从写入内存到发送数据的步骤，严格来讲这其实属于netty的范畴，因为netty写入数据也是写入内存到发送数据，只是grpc按照需求实现了一遍，包括http2header，这里对于http2header其实并没有讲到很细，但是其实http2header也经历了写入内存到发送数据的步骤，只是http2header的写入使用的就是netty默认的逻辑，总结起来就是消息体的写入，grpc有自己的实现，而对于http2header的写入，使用的就是netty默认的逻辑

本站总访问量：次本站总访问数：人
本站由 Peng Tong 使用 Stellar 1.29.1 主题创建。
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议，转载请注明出处。