|
根据rocketmq的模块设计，其通信相关的代码放在源码包下的rocketmq-remoting模块。主要内容包括了编解码处理，使用了nety框架对接收发送消息的处理等。其类图见下：  在这里插入图片描述
其中，以RemotingService为最上层接口,提供了三个接口： void start();
void shutdown();
void registerRPCHook(RPCHook rpcHook);
RemotingClient和RemotingServer都继承了RemotingService接口, 并增加了自己特有的接口.NettyRemotingClient和NettyRemotingServer分别实现了RemotingClient和RemotingServer, 并且都继承了NettyRemotingAbstract类. NettyRemotingAbstract这个抽象类包含了很多公共数据处理，也包含了很多重要的数据结构, 这个稍后介绍. 1、协议设计和编解码1.1 协议设计rocketmq的协议如下  在协议格式这里插入图片描述
从上面可以看出，其总长度是4 4 消息头长度 消息体长度。 1.2消息的编码以rocketmq给的关于remoting的test调试入手，具体类是  在这里插入图片描述
以同步通信为例 上面的例子中， requestHeader实现了CommandCustomHeader接口，即requestHeader是我们的消息头部信息。然后，以requestHeader为参，建立RemoteCommand消息。 public static RemotingCommand createRequestCommand(int code, CommandCustomHeader customHeader) {
RemotingCommand cmd = new RemotingCommand();
cmd.setCode(code);
cmd.customHeader = customHeader;
setCmdVersion(cmd);
return cmd;
}
其中，RemoteCommand是rocketmq中传输信息的消息定义体。其成员变量定义如下，其中extFields可以存储用户的键值对信息： 在test之前，会先启动remotingServer 服务端和remotingClient 客户端。 @BeforeClass
public static void setup() throws InterruptedException {
remotingServer = createRemotingServer();
remotingClient = createRemotingClient();
}
上述的服务端和客户端都一netty为基础。首先看客户端的启动createRemotingClient。客户端的启动之前，会先定义一些线程池中，创建线程如何定义等，然后调用start 其中，编码工作在new NettyEncoder()中。查看其定义： @Override
public void encode(ChannelHandlerContext ctx, RemotingCommand remotingCommand, ByteBuf out)
throws Exception {
try {
ByteBuffer header = remotingCommand.encodeHeader();
out.writeBytes(header);
byte[] body = remotingCommand.getBody();
if (body != null) {
out.writeBytes(body);
}
} catch (Exception e) {
log.error('encode exception, ' RemotingHelper.parseChannelRemoteAddr(ctx.channel()), e);
if (remotingCommand != null) {
log.error(remotingCommand.toString());
}
RemotingUtil.closeChannel(ctx.channel());
}
}
可以看到，上面主要完成了消息头和消息体的写入。其中消息体本身就是byte[]数组，不需要多做关注。重点勘察remotingCommand.encodeHeader();方法。 public ByteBuffer encodeHeader(final int bodyLength) {
// 1> header length size
int length = 4;
// 2> header data length
byte[] headerData;
//这里是重点，完成了消息头的编码
headerData = this.headerEncode();
length = headerData.length;
// 3> body data length
length = bodyLength;
ByteBuffer result = ByteBuffer.allocate(4 length - bodyLength);
// length 写入length
result.putInt(length);
// header length 写入headerlenth和序列化方式
result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));
// header data 写入头数据
result.put(headerData);
result.flip();
return result;
}
可以看到，上面的方法完成了除了body之外的消息转为bytebuf的过程。其中： 重点关注上面的 RemotingSerializable.encode(this);，其中，this指代的RemotingCommand，我们通过断点调试，看一下这个this主要包含了什么内容： RemotingCommand [code=0, language=JAVA, version=0, opaque=0, flag(B)=0, remark=null, extFields={count=1, messageTitle=Welcome}, serializeTypeCurrentRPC=JSON]
可以看到，里面内容是RemotingCommand 的相关信息，其中包括了我们自己定义的count和messageTitle信息。 总结rocketmq的编码： 消息体需要用户自己转为byte[]数组，进行传输。而消息头，是rocketmq来完成序列化和转为byte[] 数组操作。这样的设计，应该是考虑到RemotingCommand的很多设置是默认的，但又是必须的，由系统来完成消息头的序列化操作。 1.3 消息的解码消息的解码定义于NettyDecoder中， @Override
public Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
ByteBuf frame = null;
try {
frame = (ByteBuf) super.decode(ctx, in);
if (null == frame) {
return null;
}
ByteBuffer byteBuffer = frame.nioBuffer();
return RemotingCommand.decode(byteBuffer);
} catch (Exception e) {
log.error('decode exception, ' RemotingHelper.parseChannelRemoteAddr(ctx.channel()), e);
RemotingUtil.closeChannel(ctx.channel());
} finally {
if (null != frame) {
frame.release();
}
}
return null;
}
其中，解码操作位于 RemotingCommand.decode(byteBuffer); private static RemotingCommand headerDecode(byte[] headerData, SerializeType type) {
switch (type) {
//默认采用json方式将字符串转为RemotingCommand类
case JSON:
RemotingCommand resultJson = RemotingSerializable.decode(headerData, RemotingCommand.class);
resultJson.setSerializeTypeCurrentRPC(type);
return resultJson;
case ROCKETMQ:
RemotingCommand resultRMQ = RocketMQSerializable.rocketMQProtocolDecode(headerData);
resultRMQ.setSerializeTypeCurrentRPC(type);
return resultRMQ;
default:
break;
}
return null;
}
2 .rocketmq的通信流程2.1同步发送2.1.1 客户端流程真正的发送和接收数据在下面这一行：其中，为了调试方便，把时间增加到10s。即如果10s内收不到返回的数据，就报错 @Override
public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
long beginStartTime = System.currentTimeMillis();
final Channel channel = this.getAndCreateChannel(addr);
if (channel != null && channel.isActive()) {
//调用之前的操作
try {
if (this.rpcHook != null) {
this.rpcHook.doBeforeRequest(addr, request);
}
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeoutMillis < costTime) {
throw new RemotingTimeoutException('invokeSync call timeout');
}
//真正的调用在于这里
RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
//结果返回后，如果有相关操作，则执行
if (this.rpcHook != null) {
this.rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
}
return response;
} catch (RemotingSendRequestException e) {
log.warn('invokeSync: send request exception, so close the channel[{}]', addr);
this.closeChannel(addr, channel);
throw e;
} catch (RemotingTimeoutException e) {
if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
this.closeChannel(addr, channel);
log.warn('invokeSync: close socket because of timeout, {}ms, {}', timeoutMillis, addr);
}
log.warn('invokeSync: wait response timeout exception, the channel[{}]', addr);
throw e;
}
} else {
this.closeChannel(addr, channel);
throw new RemotingConnectException(addr);
}
}
再看invokeSyncImpl的定义 可以看到，我们是从ResponseFuture中取的结果，那么ResponseFuture的结果又从哪来的呢？ private final int opaque;
private final Channel processChannel;
private final long timeoutMillis;
private final InvokeCallback invokeCallback;
private final long beginTimestamp = System.currentTimeMillis();
private final CountDownLatch countDownLatch = new CountDownLatch(1);
private final SemaphoreReleaseOnlyOnce once;
private final AtomicBoolean executeCallbackOnlyOnce = new AtomicBoolean(false);
private volatile RemotingCommand responseCommand;
private volatile boolean sendRequestOK = true;
private volatile Throwable cause;
ResponseFuture类中有一些用于控制多线程的工具类，比如CountDownLatch ，Semaphore等。 public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
//从收到的数据中找到opaque，
final int opaque = cmd.getOpaque();
//从responseTable中找到此标识号的ResponseFuture
final ResponseFuture responseFuture = responseTable.get(opaque);
if (responseFuture != null) {
//把结果存入responseFuture
responseFuture.setResponseCommand(cmd);
//处理完了，移除
responseTable.remove(opaque);
if (responseFuture.getInvokeCallback() != null) {
executeInvokeCallback(responseFuture);
} else {
//好像和 responseFuture.setResponseCommand(cmd);是一样的
responseFuture.putResponse(cmd);
//异步时候有用
responseFuture.release();
}
} else {
log.warn('receive response, but not matched any request, ' RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
log.warn(cmd.toString());
}
}
上面方法中，客户端收到数据后，会将结果存入responseFuture中，而在我们前面的分析中可以看到，客户端发送完消息后，会在一定的时间之后，从responseFuture去取这个结果。 2.1.1 服务端流程在单元测试中，服务端的建立方法如下： 其中registerProcessor @Override
public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
//运行processor的线程池
ExecutorService executorThis = executor;
if (null == executor) {
executorThis = this.publicExecutor;
}
Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<NettyRequestProcessor, ExecutorService>(processor, executorThis);
//把requestCode与处理方法做成键值对，存入processorTable中
this.processorTable.put(requestCode, pair);
}
与客户端类似，服务端收到数据后，也会进行处理，流程不再说，其处理方法如下 2.2 单向(oneway) public void invokeOnewayImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
//表明是单向发送
request.markOnewayRPC();
//semaphoreOneway用于控制发送顺序，
//semaphoreOneway的默认许可是65535，每次发送前获取一次许可（许可-1），发送完成之后许可 1
boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
if (acquired) {
final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
try {
channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture f) throws Exception {
once.release();
if (!f.isSuccess()) {
log.warn('send a request command to channel <' channel.remoteAddress() '> failed.');
}
}
});
} catch (Exception e) {
once.release();
log.warn('write send a request command to channel <' channel.remoteAddress() '> failed.');
throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
}
} else {
if (timeoutMillis <= 0) {
throw new RemotingTooMuchRequestException('invokeOnewayImpl invoke too fast');
} else {
String info = String.format(
'invokeOnewayImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d',
timeoutMillis,
this.semaphoreOneway.getQueueLength(),
this.semaphoreOneway.availablePermits()
);
log.warn(info);
throw new RemotingTimeoutException(info);
}
}
}
2.3 异步调用异步调用与同步调用流程大体类似， /**
* Execute callback in callback executor. If callback executor is null, run directly in current thread
*/
private void executeInvokeCallback(final ResponseFuture responseFuture) {
boolean runInThisThread = false;
ExecutorService executor = this.getCallbackExecutor();
if (executor != null) {
try {
executor.submit(new Runnable() {
@Override
public void run() {
try {
responseFuture.executeInvokeCallback();
} catch (Throwable e) {
log.warn('execute callback in executor exception, and callback throw', e);
} finally {
responseFuture.release();
}
}
});
} catch (Exception e) {
runInThisThread = true;
log.warn('execute callback in executor exception, maybe executor busy', e);
}
} else {
runInThisThread = true;
}
if (runInThisThread) {
try {
responseFuture.executeInvokeCallback();
} catch (Throwable e) {
log.warn('executeInvokeCallback Exception', e);
} finally {
responseFuture.release();
}
}
}
上面的operationComplete就是我们在单元测试类中，定义的 remotingClient.invokeAsync('localhost:8888', request, 1000 * 3, new InvokeCallback() {
@Override
public void operationComplete(ResponseFuture responseFuture) {
latch.countDown();
System.out.println('latch.countDown()运行');
assertTrue(responseFuture != null);
assertThat(responseFuture.getResponseCommand().getLanguage()).isEqualTo(LanguageCode.JAVA);
assertThat(responseFuture.getResponseCommand().getExtFields()).hasSize(2);
}
});
© 著作权归作者所有,转载或内容合作请联系作者
|
|
|
来自： liang1234_ > 《rocketMq》
