Netty框架问题记录1--多线程下批量发送消息导致消息被覆盖

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

业务背景

项目是基于Netty实现的实时课堂项目，课堂中老师需要对试卷进行讲解，则老师向服务器发送一个打开试卷信息的请求，服务器获取试卷信息，将试卷信息发送给所有的客户端（学生和老师）。

发送给学生的时候需要在试卷信息中加上本人得分的信息。

实现方式大致如下：

Paper paper = getPaper(paperId); // 根据试卷ID获取试卷详细信息
for(Client client : allClients){
paper.setMyScore(getMyScore(client.getUserId())); //根据userId获取本人得分
client.send(paper); //向客户端发送数据
}

结果：学生A收到的得分是学生B的得分，也就是发送给clientA的paper数据被发送给clientB的paper数据给覆盖了，因为paper对象是同一个

原因分析：

虽然发送给所有客户端的信息都是paper对象，但是是在for循环里面执行的send方法，也就是说理论上应该是clientA的send方法执行完了之后才会执行clientB的send方法，也就是说理论上应该是学生A收到的paper信息之后学生B才会收到paper信息。

所以得出的结论猜想就是send方法不是同步执行的，而是异步的。追踪代码进行分析

第四行的代码client.send(paper) 实际就是调用了Channel的writeAndFlush方法

追踪到AbstractChannel的实现如下：

@Override
public ChannelFuture writeAndFlush(Object msg) {
return pipeline.writeAndFlush(msg);
}

执行了ChannelPipeline的writeAndFlush方法，跟踪实现类DefaultChannelPipeline的实现如下：

@Override
public final ChannelFuture writeAndFlush(Object msg) {
return tail.writeAndFlush(msg);
}

执行的是ChannelHandlerContext的writeAndFlush方法，跟踪实现类AbstractChannelHandlerContext实现如下：

@Override
public ChannelFuture writeAndFlush(Object msg) {
return writeAndFlush(msg, newPromise());
}

执行了内部的writeAndFlush方法，继续跟踪如下：

@Override
public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
if (msg == null) {
throw new NullPointerException("msg");
} if (isNotValidPromise(promise, true)) {
ReferenceCountUtil.release(msg);
// cancelled
return promise;
} write(msg, true, promise); return promise;
}

write方法如下：

private void write(Object msg, boolean flush, ChannelPromise promise) {
AbstractChannelHandlerContext next = findContextOutbound();
final Object m = pipeline.touch(msg, next);
EventExecutor executor = next.executor();
if (executor.inEventLoop()) { //判断当前线程是否是EventLoop线程
if (flush) {
next.invokeWriteAndFlush(m, promise);
} else {
next.invokeWrite(m, promise);
}
} else {
AbstractWriteTask task;
if (flush) {
task = WriteAndFlushTask.newInstance(next, m, promise);
} else {
task = WriteTask.newInstance(next, m, promise);
}
safeExecute(executor, task, promise, m);
}
}