应用：新美大push服务-长连通道sailfish
日推送消息：180亿
QPS峰值: 35W
最大实时在线用户：2200W
push服务简单结构为
客户端sdk<=>长连通道<=>pushServer
1.客户端sdk: 负责提供客户客户端收发push的api
2.长连通道：负责维持海量客户端连接
3.pushServer：负责给业务方提供收发push的rpc服务，与长连通道通过tcp连接，自定义协议，具体的push服务设计另起文章
首先依据这篇文章把push长连通道应用的jvm参数调到最优，见海量连接服务端jvm参数调优杂记, 剩下的都是这篇文章之后所发生
一
2016年9月2号6:00 左右陆续收到两台机器的报警，上去看一下cat监控
发现 在凌晨4:11分左右，这台机器cms old区域到达old gc阀值1525M（old区域设置为2048M, -XX:CMSInitiatingOccupancyFraction=70，所以阀值为1433M，前一分钟为1428.7M），于是进行old gc，结果进行一次old gc之后，啥也没回收掉，接下来一次次old gc，old区不减反增，甚是诡异！
在4:10:29开始频繁old gc(其实这是第二次old gc了，之前已经有过一次，不过可以忽略，我就拿这次来分析)，发现old gc过后，old区域大小基本没变，所以这个时候可以断定old区里面肯定有一直被引用的对象，猜测为缓存之类的对象
二
使用 jmap -dump:live,format=b,file=xxx [pid] 先触发一次gc再dump
重点关注这台10.32.145.237
dump 的时候，花了long long的时间,为了不影响线上引用，遂放弃。。。
三
9月3号早上又发现old gc，于是连忙起床去dump内存，总内存为1.8G，MAT载入分析
光这两个家伙就占据了71.24%，其他的可以忽略不计
然后看到NioSocketChannel这个家伙，对应着某条TCP连接，于是追根溯源，找到这条连接对应的机器
然后去cmdb里面一查
发现是pushServer的机器。长连通道服务器是用netty实现，自带缓冲区，对外连接着海量的客户端，将海量用户的请求转发给pushServer，而pushServer是BIO实现，无IO缓冲区，当pushServer的TCP缓冲区满了之后，TCP滑动窗口为0，那么长连服务器发送给这台机器的消息netty就一直会保存在自带的缓冲区ChannelOutBoundBuffer里，撑大old区。接下来需要进一步验证
四
9月5号早上，来公司验证，跑到10.12.22.253这台机器看一下tcp底层缓冲区的情况
发现tcp发送队列积压了这么多数据没发出去，这种情况发生的原因是接收方来不及处理，接收方的接收队列里面数据积压，于是导致发送方发送不出去，接下来就跑到接收方机器上看下tcp的接收队列
果不其然，三台机器接受队列都撑得很大，到这里，问题基本排查出来了，结论是接收方处理速度过慢导致发送方积压消息过多，netty会把要发送的消息保存在ChannelOutboundBuffer里面，随着积压的消息越来越多，导致old区域逐渐扩大，最终old gc，然而这些消息对象都是live的，因此回收不掉，所以频繁old gc
五
9月5号下午
考虑到pushServer改造nio需要一段时间，长连通道这边又无法忍受频繁old gc而不得不重启应用，于是在通道端做了一点更改，在选择pushServer写的时候，只选择可写的Channel
public ChannelGroupFuture writeAndFlushRandom(Object message) {
final int size = super.size();
if (size <= 0) {
return super.writeAndFlush(message);
}
return super.writeAndFlush(message, new ChannelMatcher() {
private int index = 0;
private int matchedIndex = random.nextInt(size());
@Override
public boolean matches(Channel channel) {
return matchedIndex == index++ && channel.isWritable();
}
});
}
以上&& channel.isWritable()为新添加代码，追踪一下isWritable方法的实现，最终是调用到ChannelOutboundBuffer的isWritable方法
AbstractChannel.java
@Override
public boolean isWritable() {
ChannelOutboundBuffer buf = unsafe.outboundBuffer();
return buf != null && buf.isWritable();
}
ChannelOutboundBuffer.java
public boolean isWritable() {
return unwritable == 0;
}
而unwritable这个field是在这里被确定
ChannelOutboundBuffer.java
private void incrementPendingOutboundBytes(long size, boolean invokeLater) {
if (size == 0) {
return;
}
long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, size);
if (newWriteBufferSize >= channel.config().getWriteBufferHighWaterMark()) {
setUnwritable(invokeLater);
}
}
其中channel.config().getWriteBufferHighWaterMark()返回的field是ChannelConfig里面对应的writeBufferHighWaterMark，可以看到，默认值为64K, 表示如果你在写之前调用调用isWriteable方法，netty最多给你缓存64K的数据, 否则，缓存就一直膨胀
DefaultChannelConfig.java
private volatile int writeBufferHighWaterMark = 64 * 1024;
由此可见，Channel可写至少是TCP缓冲区+netty缓冲区(默认64K)都没有写满, 我这边的做法就是当某个Channel写满之后，就放弃这条Channel，随机选择其他的Channel。
六
改造完之后，观察了一个多礼拜，old区域已缓慢稳定增长，达到预期效果
可以发现，每次old区域都是1M左右的增长
另外一个问题：每次olg gc的时候重启机器，瞬间异常井喷
重启三次，三次异常，结合前面的ChannelOutboundBuffer，不难分析，这些写失败的都是之前被堵塞的buffer，重启之后，关闭与pushServer的连接，进入到如下方法
AbstractChannel.java
public final void close(final ChannelPromise promise) {
if (!promise.setUncancellable()) {
return;
}
if (outboundBuffer == null) {
// Only needed if no VoidChannelPromise.
if (!(promise instanceof VoidChannelPromise)) {
// This means close() was called before so we just register a listener and return
closeFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
promise.setSuccess();
}
});
}
return;
}
if (closeFuture.isDone()) {
// Closed already.
safeSetSuccess(promise);
return;
}
final boolean wasActive = isActive();
final ChannelOutboundBuffer buffer = outboundBuffer;
outboundBuffer = null; // Disallow adding any messages and flushes to outboundBuffer.
Executor closeExecutor = closeExecutor();
if (closeExecutor != null) {
closeExecutor.execute(new OneTimeTask() {
@Override
public void run() {
try {
// Execute the close.
doClose0(promise);
} finally {
// Call invokeLater so closeAndDeregister is executed in the EventLoop again!
invokeLater(new OneTimeTask() {
@Override
public void run() {
// Fail all the queued messages
buffer.failFlushed(CLOSED_CHANNEL_EXCEPTION, false);
buffer.close(CLOSED_CHANNEL_EXCEPTION);
fireChannelInactiveAndDeregister(wasActive);
}
});
}
}
});
} else {
try {
// Close the channel and fail the queued messages in all cases.
doClose0(promise);
} finally {
// Fail all the queued messages.
buffer.failFlushed(CLOSED_CHANNEL_EXCEPTION, false);
buffer.close(CLOSED_CHANNEL_EXCEPTION);
}
if (inFlush0) {
invokeLater(new OneTimeTask() {
@Override
public void run() {
fireChannelInactiveAndDeregister(wasActive);
}
});
} else {
fireChannelInactiveAndDeregister(wasActive);
}
}
}
程序进入到outboundBuffer.failFlushed(CLOSED_CHANNEL_EXCEPTION, false);
然后进入到ChannelOutboundBuffer的failFlushed方法
void failFlushed(Throwable cause, boolean notify) {
// Make sure that this method does not reenter. A listener added to the current promise can be notified by the
// current thread in the tryFailure() call of the loop below, and the listener can trigger another fail() call
// indirectly (usually by closing the channel.)
//
// See https://github.com/netty/netty/issues/1501
if (inFail) {
return;
}
try {
inFail = true;
for (;;) {
if (!remove0(cause, notify)) {
break;
}
}
} finally {
inFail = false;
}
}
这里的for循环导致remove0 会遍历Entry缓存对象链表
private boolean remove0(Throwable cause, boolean notifyWritability) {
Entry e = flushedEntry;
if (e == null) {
clearNioBuffers();
return false;
}
Object msg = e.msg;
ChannelPromise promise = e.promise;
int size = e.pendingSize;
removeEntry(e);
if (!e.cancelled) {
// only release message, fail and decrement if it was not canceled before.
ReferenceCountUtil.safeRelease(msg);
safeFail(promise, cause);
decrementPendingOutboundBytes(size, false, notifyWritability);
}
// recycle the entry
e.recycle();
return true;
}
private void removeEntry(Entry e) {
if (-- flushed == 0) {
// processed everything
flushedEntry = null;
if (e == tailEntry) {
tailEntry = null;
unflushedEntry = null;
}
} else {
// 指针指向下一个待删除的缓存
flushedEntry = e.next;
}
}
直到所有的缓存对象都被remove掉,remove0 每调用一次都会调用一次safeFail(promise, cause)方法，
private static void safeFail(ChannelPromise promise, Throwable cause) {
if (!(promise instanceof VoidChannelPromise) && !promise.tryFailure(cause)) {
logger.warn("Failed to mark a promise as failure because it's done already: {}", promise, cause);
}
}
然后进入到
DefaultPromise.java
@Override
public boolean tryFailure(Throwable cause) {
if (setFailure0(cause)) {
notifyListeners();
return true;
}
return false;
}
DefaultPromise.java
static void notifyListener0(Future future, GenericFutureListener l) {
try {
l.operationComplete(future);
} catch (Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("An exception was thrown by " + l.getClass().getName() + ".operationComplete()", t);
}
}
}
最终一个future回调，调回到用户方法
@Override
protected void channelRead0(ChannelHandlerContext ctx, TransferFromSdkDataPacket msg) throws Exception {
TransferToPushServerDataPacket dataPacket = new TransferToPushServerDataPacket();
dataPacket.setVersion(Constants.PUSH_SERVER_VERSION);
dataPacket.setData(msg.getData());
dataPacket.setConnectionId(ctx.channel().attr(AttributeKeys.CONNECTION_ID).get());
final long startTime = System.nanoTime();
try {
ChannelGroupFuture channelFutures = pushServerChannels.writeAndFlushRandom(dataPacket);
channelFutures.addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
if (future.isSuccess()) {
CatUtil.logTransaction(startTime, null, CatTransactions.TransferToPushServer, CatTransactions.TransferToPushServer);
} else {
Channel channel = (Channel) ((ChannelGroupFuture) future).group().toArray()[0];
final String pushServer = channel.remoteAddress().toString();
TransferToPushServerException e = new TransferToPushServerException(String.format("pushServer: %s", pushServer), future.cause());
CatUtil.logTransaction(new CatUtil.CatTransactionCallBack() {
@Override
protected void beforeComplete() {
Cat.logEvent(CatEvents.WriteToPushServerError, pushServer);
}
}, startTime, e, CatTransactions.TransferToPushServer, CatTransactions.TransferToPushServer);
}
}
});
} catch (Exception e) {
CatUtil.logTransaction(startTime, e, CatTransactions.TransferToPushServer, CatTransactions.TransferToPushServer);
}
}
而在用户方法里面，我们包装了一下自定义异常，喷到cat，导致瞬间TransferToPushServerException飙高
如果你觉得看的不过瘾，想系统学习Netty原理，那么你一定不要错过我的Netty源码分析系列视频：https://coding.imooc.com/class/230.html
