当点击CAT的查看LogViews时出现Sorry, the message is not there. It could be missing or archived.
这时候出现这种问题会一头雾水，去github上查看貌似也没有给出明确答复。
这里根据自己的猜想以及源码角度的查看来定位问题。
首先咨询CAT维护相关人员得到一个非常重要的信息就是消息编号中的小时数不同
项目名-ac13bd78-430207-91
项目名-ac13bd78-430208-91
第二段代表的是IP，第三段代表小时数，第四段代表当前自增数
举例:
从LogViews中的消息编号列表中发现当前小时如果属于430207的话,是可以打开的,非430207的话就是消息丢失。
然后从CAT的state报表中发现有一列消息丢失数据:
两台机器时钟不准导致消息存储丢失 这个场景用于Pigeon，服务端id是由客户端产生，客户端和服务端时钟差2小时，会导致存储丢失
猜想这一列是不是统计了我刚好消息丢失的数据。
有了这些线索，我们开始假设和验证！
从客户端编号的消息编号生成开始入手。
我的客户端版本是 2.0
源码查看
- 发送消息的时候会判断该消息是否是EVENT消息,如果有则放入m_atomicTrees对象中
TcpSocketSender.java
public void send(MessageTree tree) {
if (isAtomicMessage(tree)) {
boolean result = m_atomicTrees.offer(tree, m_manager.getSample());
if (!result) {
logQueueFullInfo(tree);
}
} else {
boolean result = m_queue.offer(tree, m_manager.getSample());
if (!result) {
logQueueFullInfo(tree);
}
}
}
- 一旦放入m_atomicTrees对象时,则会被一个单独监控的线程检测到
public class MergeAtomicTask implements Task {
@Override
public String getName() {
return "merge-atomic-task";
}
@Override
public void run() {
while (true) {
// 时刻监控这个队列,一旦有并且是当前小时的消息则会满足
if (shouldMerge(m_atomicTrees)) {
MessageTree tree = mergeTree(m_atomicTrees);
boolean result = m_queue.offer(tree);
if (!result) {
logQueueFullInfo(tree);
}
} else {
try {
Thread.sleep(5);
} catch (InterruptedException e) {
break;
}
}
}
}
@Override
public void shutdown() {
}
}
mergeTree这个方法有点重要,其实就是将当前的消息进行合并，为什么需要合并？
我猜的话应该是在同一个消息树中，每条消息都对应了一个消息编号，但是对于消息树的编号来说只要第一条消息的消息编号就能够定位到，而第二条往后走的消息这个编号根本就没用了,但是不想浪费了，放入消息编号队列中，为下一条消息树的编号生成所用。
private MessageTree mergeTree(MessageQueue trees) {
int max = MAX_CHILD_NUMBER;
DefaultTransaction t = new DefaultTransaction("_CatMergeTree", "_CatMergeTree", null);
// 先获取消息树中的第一条消息
MessageTree first = trees.poll();
t.setStatus(Transaction.SUCCESS);
t.setCompleted(true);
t.addChild(first.getMessage());
t.setTimestamp(first.getMessage().getTimestamp());
long lastTimestamp = 0;
long lastDuration = 0;
while (max >= 0) {
// 注意这里是从第二条开始
MessageTree tree = trees.poll();
if (tree == null) {
t.setDurationInMillis(lastTimestamp - t.getTimestamp() + lastDuration);
break;
}
lastTimestamp = tree.getMessage().getTimestamp();
if (tree.getMessage() instanceof DefaultTransaction) {
lastDuration = ((DefaultTransaction) tree.getMessage()).getDurationInMillis();
} else {
lastDuration = 0;
}
t.addChild(tree.getMessage());
// 这里非常关键，会将本次产生的id编号重新放入生成的队列中。
m_factory.reuse(tree.getMessageId());
max--;
}
((DefaultMessageTree) first).setMessage(t);
return first;
}
// D:\lib\maven\com\dianping\cat\cat-client\2.0.0\cat-client-2.0.0-sources.jar!\com\dianping\cat\message\internal\MessageIdFactory.java
// m_factory.reuse(tree.getMessageId()); 对应的实现
public void reuse(String id) {
m_reusedIds.offer(id);
}
接下来我们只要看它是如何拿id的
MessageIdFactory.java
public String getNextId() {
// 先从队列里面拿到,这里和上面生成的相关,如果有直接返回。
String id = m_reusedIds.poll();
if (id != null) {
return id;
} else {
long timestamp = getTimestamp();
if (timestamp != m_timestamp) {
m_index = new AtomicInteger(0);
m_timestamp = timestamp;
}
int index = m_index.getAndIncrement();
StringBuilder sb = new StringBuilder(m_domain.length() + 32);
sb.append(m_domain);
sb.append('-');
sb.append(m_ipAddress);
sb.append('-');
sb.append(timestamp);
sb.append('-');
sb.append(index);
return sb.toString();
}
}
这里就会出现一个小问题,如果当前小时的id没有拿完，下一个小时来拿id的时候发现还有，则会从队列里面继续获取生成的id编号，生成的消息树发送到服务端，但问题在于，该编号却是上个小时残留的。
这时候服务端是以小时为key存储的,存储的时候会发现这个编号是上一个小时的,则会直接丢弃。从state中的
两台机器时钟不准导致消息存储丢失 | 这个场景用于Pigeon，服务端id是由客户端产生，客户端和服务端时钟差2小时，会导致存储丢失
中查看到!
这部分的源码在 : TcpSocketReceiver.MessageDecoder.decode()中体现出来，最终的实现类是RealtimeConsumer.consume方法
public void consume(MessageTree tree) {
long timestamp = tree.getMessage().getTimestamp();
Period period = m_periodManager.findPeriod(timestamp);
if (period != null) {
// 将消息树交给桶处理 , 然后这里会放到队列然后交给另一个PeriodTask线程去处理
period.distribute(tree);
} else {
m_serverStateManager.addNetworkTimeError(1);
}
}
PeriodTask.run -> AbstractMessageAnalyzer.analyze -> DumpAnalyzer.process -> processWithStorage 处理
DumpAnalyzer.java - 关键代码
public void process(MessageTree tree) {
try {
// 根据消息编号做解析
MessageId messageId = MessageId.parse(tree.getMessageId());
if (!shouldDiscard(messageId)) {
// 这里会获取消息编号的第三段作为参数messageId.getHour()
processWithStorage(tree, messageId, messageId.getHour());
}
} catch (Exception ignored) {
}
}
private void processWithStorage(MessageTree tree, MessageId messageId, int hour) {
// 这个桶是以当前小时为单位 也就是消息编号的第三段
MessageDumper dumper = m_dumperManager.find(hour);
tree.setFormatMessageId(messageId);
// 这里根据消息编号发现消息编号匹配不到
if (dumper != null) {
dumper.process(tree);
} else {
// 然后state那里就会多一条数据
m_serverStateManager.addPigeonTimeError(1);
}
}
ServerStatistic.Statistic.m_pigeonTimeError
但这个时候是误导了用户，实际上是消息队列中残留了上个小时的消息生成的编号导致的。
解决方案:
客户端升级到3.0吧。它已经把队列去掉了,每次获取当前时间戳，来生成编号。
网友评论