之前讲了 cat-client 进行cat埋点上报，那么上报给谁呢？以及后续故事如何？让我们来看看 cat-consumer 是如何接收处理的？
由cat-client发送数据，cat-consumer进行接收请求处理，开始了处理问题之旅！
首先，让我们来回顾一下 TcpSocketSender 是如何发送数据的：
// TcpSocketSender 往channel中写入数据，此处有兴趣的同学可以延伸下 netty 的源码！
private void sendInternal(MessageTree tree) {
ChannelFuture future = m_manager.channel();
ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(10 * 1024); // 10K m_codec.encode(tree, buf); int size = buf.readableBytes();
Channel channel = future.channel(); // 以 ByteBuf 形式发送数据
channel.writeAndFlush(buf);
// 更新统计数据
if (m_statistics != null) {
m_statistics.onBytes(size);
}
}
// TcpSocketReceiver, 接收发送过来的数据，默认端口 2280, 注册服务，线上为分布式部署，即为接口调用式。
public void init() {
try {
startServer(m_port);
} catch (Throwable e) {
m_logger.error(e.getMessage(), e);
}
}
public synchronized void startServer(int port) throws InterruptedException {
boolean linux = getOSMatches("Linux") || getOSMatches("LINUX");
int threads = 24;
ServerBootstrap bootstrap = new ServerBootstrap();
m_bossGroup = linux ? new EpollEventLoopGroup(threads) : new NioEventLoopGroup(threads);
m_workerGroup = linux ? new EpollEventLoopGroup(threads) : new NioEventLoopGroup(threads);
bootstrap.group(m_bossGroup, m_workerGroup);
bootstrap.channel(linux ? EpollServerSocketChannel.class : NioServerSocketChannel.class);
// 添加处理handler, 进行请求逻辑处理
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 此处仅为一个解码器，实际功能在该解码器中完成
pipeline.addLast("decode", new MessageDecoder());
}
});
bootstrap.childOption(ChannelOption.SO_REUSEADDR, true);
bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
try {
m_future = bootstrap.bind(port).sync();
m_logger.info("start netty server!");
} catch (Exception e) {
m_logger.error("Started Netty Server Failed:" + port, e);
}
}
// 消息解码器，并处理具体业务逻辑，先确认数据已上传完成，再进行逻辑处理
public class MessageDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
if (buffer.readableBytes() < 4) {
return;
}
buffer.markReaderIndex();
int length = buffer.readInt();
buffer.resetReaderIndex();
if (buffer.readableBytes() < length + 4) {
return;
}
try {
if (length > 0) {
ByteBuf readBytes = buffer.readBytes(length + 4);
readBytes.markReaderIndex();
readBytes.readInt();
// 消息解码，获取头信息，消息体
DefaultMessageTree tree = (DefaultMessageTree) m_codec.decode(readBytes);
readBytes.resetReaderIndex();
tree.setBuffer(readBytes);
// 交由handler处理实际逻辑
m_handler.handle(tree);
m_processCount++;
long flag = m_processCount % CatConstants.SUCCESS_COUNT;
if (flag == 0) {
m_serverStateManager.addMessageTotal(CatConstants.SUCCESS_COUNT);
}
} else {
// client message is error
buffer.readBytes(length);
}
} catch (Exception e) {
m_serverStateManager.addMessageTotalLoss(1);
m_logger.error(e.getMessage(), e);
}
}
}
// 消息解码，为后续处理提供基础设施
//PlainTextMessageCodec.decode() 消息解码
@Override
public MessageTree decode(ByteBuf buf) {
MessageTree tree = new DefaultMessageTree(); // 使用一个默认消息树，用于接收消息
decode(buf, tree);
return tree;
} @Override
public void decode(ByteBuf buf, MessageTree tree) {
Context ctx = m_ctx.get().setBuffer(buf); // 解析头信息
decodeHeader(ctx, tree); // 解析消息体
if (buf.readableBytes() > 0) {
decodeMessage(ctx, tree);
}
}
// 解析头信息，以tab='\t' 和 lf='\n', 进行分割
protected void decodeHeader(Context ctx, MessageTree tree) {
BufferHelper helper = m_bufferHelper;
String id = helper.read(ctx, TAB);
String domain = helper.read(ctx, TAB);
String hostName = helper.read(ctx, TAB);
String ipAddress = helper.read(ctx, TAB);
String threadGroupName = helper.read(ctx, TAB);
String threadId = helper.read(ctx, TAB);
String threadName = helper.read(ctx, TAB);
String messageId = helper.read(ctx, TAB);
String parentMessageId = helper.read(ctx, TAB);
String rootMessageId = helper.read(ctx, TAB);
String sessionToken = helper.read(ctx, LF); if (VERSION.equals(id)) {
tree.setDomain(domain);
tree.setHostName(hostName);
tree.setIpAddress(ipAddress);
tree.setThreadGroupName(threadGroupName);
tree.setThreadId(threadId);
tree.setThreadName(threadName);
tree.setMessageId(messageId);
tree.setParentMessageId(parentMessageId);
tree.setRootMessageId(rootMessageId);
tree.setSessionToken(sessionToken);
} else {
throw new RuntimeException(String.format("Unrecognized id(%s) for plain text message codec!", id));
}
}
// 解析消息体
protected void decodeMessage(Context ctx, MessageTree tree) {
Stack<DefaultTransaction> stack = new Stack<DefaultTransaction>();
Message parent = decodeLine(ctx, null, stack); tree.setMessage(parent); // 循环读取消息体，直到读取完成
while (ctx.getBuffer().readableBytes() > 0) {
Message message = decodeLine(ctx, (DefaultTransaction) parent, stack); if (message instanceof DefaultTransaction) {
parent = message;
} else {
break;
}
}
}
// 解析内容栈出来
protected Message decodeLine(Context ctx, DefaultTransaction parent, Stack<DefaultTransaction> stack) {
BufferHelper helper = m_bufferHelper;
byte identifier = ctx.getBuffer().readByte();
String timestamp = helper.read(ctx, TAB);
String type = helper.read(ctx, TAB);
String name = helper.read(ctx, TAB); switch (identifier) {
// t: transaction 类型消息, T: pop结束, E:Event, M: Metrics, L: Trace, H: heartbeat 消息
case 't':
DefaultTransaction transaction = new DefaultTransaction(type, name, null); helper.read(ctx, LF); // get rid of line feed
transaction.setTimestamp(m_dateHelper.parse(timestamp)); if (parent != null) {
parent.addChild(transaction);
} stack.push(parent);
return transaction;
case 'A':
DefaultTransaction tran = new DefaultTransaction(type, name, null);
String status = helper.read(ctx, TAB);
String duration = helper.read(ctx, TAB);
String data = helper.read(ctx, TAB); helper.read(ctx, LF); // get rid of line feed
tran.setTimestamp(m_dateHelper.parse(timestamp));
tran.setStatus(status);
tran.addData(data); long d = Long.parseLong(duration.substring(0, duration.length() - 2));
tran.setDurationInMicros(d); if (parent != null) {
parent.addChild(tran);
return parent;
} else {
return tran;
}
case 'T':
String transactionStatus = helper.read(ctx, TAB);
String transactionDuration = helper.read(ctx, TAB);
String transactionData = helper.read(ctx, TAB); helper.read(ctx, LF); // get rid of line feed
parent.setStatus(transactionStatus);
parent.addData(transactionData); long transactionD = Long.parseLong(transactionDuration.substring(0, transactionDuration.length() - 2)); parent.setDurationInMicros(transactionD); return stack.pop();
case 'E':
DefaultEvent event = new DefaultEvent(type, name);
String eventStatus = helper.read(ctx, TAB);
String eventData = helper.read(ctx, TAB); helper.read(ctx, LF); // get rid of line feed
event.setTimestamp(m_dateHelper.parse(timestamp));
event.setStatus(eventStatus);
event.addData(eventData); if (parent != null) {
parent.addChild(event);
return parent;
} else {
return event;
}
case 'M':
DefaultMetric metric = new DefaultMetric(type, name);
String metricStatus = helper.read(ctx, TAB);
String metricData = helper.read(ctx, TAB); helper.read(ctx, LF); // get rid of line feed
metric.setTimestamp(m_dateHelper.parse(timestamp));
metric.setStatus(metricStatus);
metric.addData(metricData); if (parent != null) {
parent.addChild(metric);
return parent;
} else {
return metric;
}
case 'L':
DefaultTrace trace = new DefaultTrace(type, name);
String traceStatus = helper.read(ctx, TAB);
String traceData = helper.read(ctx, TAB); helper.read(ctx, LF); // get rid of line feed
trace.setTimestamp(m_dateHelper.parse(timestamp));
trace.setStatus(traceStatus);
trace.addData(traceData); if (parent != null) {
parent.addChild(trace);
return parent;
} else {
return trace;
}
case 'H':
DefaultHeartbeat heartbeat = new DefaultHeartbeat(type, name);
String heartbeatStatus = helper.read(ctx, TAB);
String heartbeatData = helper.read(ctx, TAB); helper.read(ctx, LF); // get rid of line feed
heartbeat.setTimestamp(m_dateHelper.parse(timestamp));
heartbeat.setStatus(heartbeatStatus);
heartbeat.addData(heartbeatData); if (parent != null) {
parent.addChild(heartbeat);
return parent;
} else {
return heartbeat;
}
default:
m_logger.warn("Unknown identifier(" + (char) identifier + ") of message: "
+ ctx.getBuffer().toString(Charset.forName("utf-8")));
throw new RuntimeException("Unknown identifier int name");
}
}
// handler 处理流程，由DefaultMessageHandler接手，安排后续工作。
// DefaultMessageHandler, 接过处理器的第一棒, 交由另一实际的consumer(RealtimeConsumer) handler处理
@Override
public void handle(MessageTree tree) {
if (m_consumer == null) {
m_consumer = lookup(MessageConsumer.class);
} try {
m_consumer.consume(tree);
} catch (Throwable e) {
m_logger.error("Error when consuming message in " + m_consumer + "! tree: " + tree, e);
}
}
// RealtimeConsumer, 进行消费数据
@Override
public void consume(MessageTree tree) {
String domain = tree.getDomain();
String ip = tree.getIpAddress(); // 进行权限检测，ip,domain
if (!m_blackListManager.isBlack(domain, ip)) {
long timestamp = tree.getMessage().getTimestamp();
Period period = m_periodManager.findPeriod(timestamp); // 找到period, 再将消息分配过去，否则算作网络异常
if (period != null) {
period.distribute(tree);
} else {
m_serverStateManager.addNetworkTimeError(1);
}
} else {
m_black++; if (m_black % CatConstants.SUCCESS_COUNT == 0) {
Cat.logEvent("Discard", domain);
}
}
}
// Period.distribute, 将消息依次取出，进行分发到队列
public void distribute(MessageTree tree) {
// 统计进行数进行加1
m_serverStateManager.addMessageTotal(tree.getDomain(), 1);
boolean success = true;
String domain = tree.getDomain();
// 将各种类型的监控数据分别取出进行处理
for (Entry<String, List<PeriodTask>> entry : m_tasks.entrySet()) {
List<PeriodTask> tasks = entry.getValue();
int length = tasks.size();
int index = 0;
boolean manyTasks = length > 1;
if (manyTasks) {
index = Math.abs(domain.hashCode()) % length;
}
PeriodTask task = tasks.get(index);
// 如果有金条消息，将task重新入队
boolean enqueue = task.enqueue(tree);
if (enqueue == false) {
if (manyTasks) {
task = tasks.get((index + 1) % length);
enqueue = task.enqueue(tree);
if (enqueue == false) {
success = false;
}
} else {
success = false;
}
}
}
if (!success) {
m_serverStateManager.addMessageTotalLoss(tree.getDomain(), 1);
}
}
// PeriodTask.enqueue, 重新入队消息，让消费线程自行消费 LinkedBlockingQueue.offer(..)
public boolean enqueue(MessageTree tree) {
boolean result = m_queue.offer(tree);
if (!result) { // trace queue overflow, 记录入队失败日志
m_queueOverflow++;
if (m_queueOverflow % (10 * CatConstants.ERROR_COUNT) == 0) {
m_logger.warn(m_analyzer.getClass().getSimpleName() + " queue overflow number " + m_queueOverflow);
}
}
return result;
}
到此，一条消费线路就完成了。
// PeriodTask 线程，作为第二个消费线路
@Override
public void run() {
try {
// 分析各消息数据，做后台消费处理
m_analyzer.analyze(m_queue);
} catch (Exception e) {
Cat.logError(e);
}
}
// 调用统一的抽象类的模板方法，由各类进行具体的 process 处理
@Override
public void analyze(MessageQueue queue) {
while (!isTimeout() && isActive()) {
MessageTree tree = queue.poll(); if (tree != null) {
try {
// 调用具体类的process
process(tree);
} catch (Throwable e) {
m_errors++; if (m_errors == 1 || m_errors % 10000 == 0) {
Cat.logError(e);
}
}
}
} // 如果出现超时或者停止动作，则把剩余队列处理完成再退出线程
while (true) {
MessageTree tree = queue.poll(); if (tree != null) {
try {
process(tree);
} catch (Throwable e) {
m_errors++; if (m_errors == 1 || m_errors % 10000 == 0) {
Cat.logError(e);
}
}
} else {
break;
}
}
}
// 超时规则，当前时间 > 开始时间+1小时+设置额外超时时间
protected boolean isTimeout() {
long currentTime = System.currentTimeMillis();
long endTime = m_startTime + m_duration + m_extraTime; return currentTime > endTime;
}
// 具体的 Anlalyzer示例: DumpAnlalyzer.process
// 具体的 Anlalyzer示例: DumpAnlalyzer.process
@Override
public void process(MessageTree tree) {
String domain = tree.getDomain(); if ("PhoenixAgent".equals(domain)) {
return;
} else {
MessageId messageId = MessageId.parse(tree.getMessageId()); if (messageId.getVersion() == 2) {
// 计算出当前时间范围，
long time = tree.getMessage().getTimestamp();
long fixedTime = time - time % (TimeHelper.ONE_HOUR);
long idTime = messageId.getTimestamp();
long duration = fixedTime - idTime; if (duration == 0 || duration == ONE_HOUR || duration == -ONE_HOUR) {
m_bucketManager.storeMessage(tree, messageId);
} else {
m_serverStateManager.addPigeonTimeError(1);
}
}
}
}
// 存储log消息到本地文件，并后续上传到hdfs
@Override
public void storeMessage(final MessageTree tree, final MessageId id) {
boolean errorFlag = true;
int hash = Math.abs((id.getDomain() + '-' + id.getIpAddress()).hashCode());
int index = (int) (hash % m_gzipThreads);
MessageItem item = new MessageItem(tree, id);
LinkedBlockingQueue<MessageItem> queue = m_messageQueues.get(index % (m_gzipThreads - 1));
boolean result = queue.offer(item); if (result) {
errorFlag = false;
} else {
if (m_last.offer(item)) {
errorFlag = false;
}
} if (errorFlag) {
m_serverStateManager.addMessageDumpLoss(1);
}
logStorageState(tree);
}
// 每1000个消息添加一个messageDump=1000
protected void logStorageState(final MessageTree tree) {
String domain = tree.getDomain();
int size = ((DefaultMessageTree) tree).getBuffer().readableBytes(); m_serverStateManager.addMessageSize(domain, size);
if ((++m_total) % CatConstants.SUCCESS_COUNT == 0) {
m_serverStateManager.addMessageDump(CatConstants.SUCCESS_COUNT);
}
}
// EventAnalyzer.process 处理event消息
@Override
public void process(MessageTree tree) {
String domain = tree.getDomain(); if (m_serverFilterConfigManager.validateDomain(domain)) {
EventReport report = m_reportManager.getHourlyReport(getStartTime(), domain, true);
Message message = tree.getMessage();
String ip = tree.getIpAddress(); if (message instanceof Transaction) {
processTransaction(report, tree, (Transaction) message, ip);
} else if (message instanceof Event) {
processEvent(report, tree, (Event) message, ip);
}
}
}
// 循环处理多个transation
private void processTransaction(EventReport report, MessageTree tree, Transaction t, String ip) {
List<Message> children = t.getChildren(); for (Message child : children) {
if (child instanceof Transaction) {
processTransaction(report, tree, (Transaction) child, ip);
} else if (child instanceof Event) {
processEvent(report, tree, (Event) child, ip);
}
}
}
// StateAnalyzer.process 对cat的机器作展示
@Override
protected void process(MessageTree tree) {
String domain = tree.getDomain(); if (m_serverFilterConfigManager.validateDomain(domain)) {
StateReport report = m_reportManager.getHourlyReport(getStartTime(), Constants.CAT, true);
String ip = tree.getIpAddress();
Machine machine = report.findOrCreateMachine(NetworkInterfaceManager.INSTANCE.getLocalHostAddress()); machine.findOrCreateProcessDomain(domain).addIp(ip);
}
}
// 所有分析线程，由 Period 进行初始化启动所有的Analyzer备用
public void start() {
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
m_logger.info(String.format("Starting %s tasks in period [%s, %s]", m_tasks.size(),
df.format(new Date(m_startTime)), df.format(new Date(m_endTime - 1))));
for (Entry<String, List<PeriodTask>> tasks : m_tasks.entrySet()) {
List<PeriodTask> taskList = tasks.getValue();
for (int i = 0; i < taskList.size(); i++) {
PeriodTask task = taskList.get(i);
task.setIndex(i);
Threads.forGroup("Cat-RealtimeConsumer").start(task);
}
}
}
// 为保证高可用，使用 ChannelManager, 专门检查channel通道是否仍然存活，如果出问题，则发起重连。
@Override
public void run() {
while (m_active) {
// make save message id index asyc
m_idfactory.saveMark();
checkServerChanged(); ChannelFuture activeFuture = m_activeChannelHolder.getActiveFuture();
List<InetSocketAddress> serverAddresses = m_activeChannelHolder.getServerAddresses(); doubleCheckActiveServer(activeFuture);
reconnectDefaultServer(activeFuture, serverAddresses); try {
Thread.sleep(10 * 1000L); // check every 10 seconds
} catch (InterruptedException e) {
// ignore
}
}
}
// 最后一个关键点也是很重要的一个点，PeriodManager, 用于滚动式处理每小时的监控数据，保存数据到磁盘
// PeriodManager, 用于滚动式处理每小时的监控数据
public class PeriodManager implements Task {
private PeriodStrategy m_strategy;
private List<Period> m_periods = new ArrayList<Period>();
private boolean m_active;
@Inject
private MessageAnalyzerManager m_analyzerManager;
@Inject
private ServerStatisticManager m_serverStateManager;
@Inject
private Logger m_logger;
public static long EXTRATIME = 3 * 60 * 1000L;
public PeriodManager(long duration, MessageAnalyzerManager analyzerManager,
ServerStatisticManager serverStateManager, Logger logger) {
m_strategy = new PeriodStrategy(duration, EXTRATIME, EXTRATIME);
m_active = true;
m_analyzerManager = analyzerManager;
m_serverStateManager = serverStateManager;
m_logger = logger;
}
private void endPeriod(long startTime) {
int len = m_periods.size();
for (int i = 0; i < len; i++) {
Period period = m_periods.get(i);
if (period.isIn(startTime)) {
period.finish();
m_periods.remove(i);
break;
}
}
}
public void init() {
long startTime = m_strategy.next(System.currentTimeMillis());
startPeriod(startTime);
}
@Override
public void run() {
while (m_active) {
try {
long now = System.currentTimeMillis();
long value = m_strategy.next(now);
if (value > 0) {
startPeriod(value);
} else if (value < 0) {
// 上个运行周期，即1小时已完成后，启用一个结束线程进行规划原来的数据
Threads.forGroup("cat").start(new EndTaskThread(-value));
}
} catch (Throwable e) {
Cat.logError(e);
}
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
break;
}
}
}
private void startPeriod(long startTime) {
long endTime = startTime + m_strategy.getDuration();
Period period = new Period(startTime, endTime, m_analyzerManager, m_serverStateManager, m_logger);
m_periods.add(period);
period.start();
}
private class EndTaskThread implements Task {
private long m_startTime;
public EndTaskThread(long startTime) {
m_startTime = startTime;
}
@Override
public void run() {
// 调用外部类的结束方法
endPeriod(m_startTime);
}
}
}
// Period.finish(), 结束
public void finish() {
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date startDate = new Date(m_startTime);
Date endDate = new Date(m_endTime - 1);
m_logger.info(String.format("Finishing %s tasks in period [%s, %s]", m_tasks.size(), df.format(startDate),
df.format(endDate)));
try {
for (Entry<String, List<PeriodTask>> tasks : m_tasks.entrySet()) {
for (PeriodTask task : tasks.getValue()) {
task.finish();
}
}
} catch (Throwable e) {
Cat.logError(e);
} finally {
m_logger.info(String.format("Finished %s tasks in period [%s, %s]", m_tasks.size(), df.format(startDate),
df.format(endDate)));
}
}
// PeriodTask.finish(), 真正处理上一周期数据
public void finish() {
try {
// 调用各自分析器的 doCheckpoint 检查，后续处理
m_analyzer.doCheckpoint(true);
// 销毁分析器, help gc
m_analyzer.destroy();
} catch (Exception e) {
Cat.logError(e);
}
}
// 举例 EventAnalyzer.doCheckpoint, 需加锁处理
@Override
public synchronized void doCheckpoint(boolean atEnd) {
if (atEnd && !isLocalMode()) {
m_reportManager.storeHourlyReports(getStartTime(), StoragePolicy.FILE_AND_DB, m_index);
} else {
m_reportManager.storeHourlyReports(getStartTime(), StoragePolicy.FILE, m_index);
}
}
// DefaultReportManager.storeHourlyReports, 存储logview, 存在统计数据到db
@Override
public void storeHourlyReports(long startTime, StoragePolicy policy, int index) {
Transaction t = Cat.newTransaction("Checkpoint", m_name);
Map<String, T> reports = m_reports.get(startTime);
ReportBucket bucket = null;
try {
t.addData("reports", reports == null ? 0 : reports.size());
if (reports != null) {
Set<String> errorDomains = new HashSet<String>();
for (String domain : reports.keySet()) {
if (!m_validator.validate(domain)) {
errorDomains.add(domain);
}
}
for (String domain : errorDomains) {
reports.remove(domain);
}
if (!errorDomains.isEmpty()) {
m_logger.info("error domain:" + errorDomains);
}
m_reportDelegate.beforeSave(reports);
if (policy.forFile()) {
bucket = m_bucketManager.getReportBucket(startTime, m_name, index);
try {
storeFile(reports, bucket);
} finally {
m_bucketManager.closeBucket(bucket);
}
}
if (policy.forDatabase()) {
storeDatabase(startTime, reports);
}
}
t.setStatus(Message.SUCCESS);
} catch (Throwable e) {
Cat.logError(e);
t.setStatus(e);
m_logger.error(String.format("Error when storing %s reports of %s!", m_name, new Date(startTime)), e);
} finally {
cleanup(startTime);
t.complete();
if (bucket != null) {
m_bucketManager.closeBucket(bucket);
}
}
}
// DefaultReportManager.storeDatabase, 存储 db
private void storeDatabase(long startTime, Map<String, T> reports) {
Date period = new Date(startTime);
String ip = NetworkInterfaceManager.INSTANCE.getLocalHostAddress();
for (T report : reports.values()) {
try {
String domain = m_reportDelegate.getDomain(report);
HourlyReport r = m_reportDao.createLocal();
r.setName(m_name);
r.setDomain(domain);
r.setPeriod(period);
r.setIp(ip);
r.setType(1);
m_reportDao.insert(r);
int id = r.getId();
byte[] binaryContent = m_reportDelegate.buildBinary(report);
HourlyReportContent content = m_reportContentDao.createLocal();
content.setReportId(id);
content.setContent(binaryContent);
m_reportContentDao.insert(content);
m_reportDelegate.createHourlyTask(report);
} catch (Throwable e) {
Cat.getProducer().logError(e);
}
}
}
// DefaultReportManager.storeFile, 存在file, xml
private void storeFile(Map<String, T> reports, ReportBucket bucket) {
for (T report : reports.values()) {
try {
String domain = m_reportDelegate.getDomain(report);
String xml = m_reportDelegate.buildXml(report);
bucket.storeById(domain, xml);
} catch (Exception e) {
Cat.logError(e);
}
}
}
总结起来就几个东西：
1. 使用netty开启高性能的接收服务;
2. 使用队列进行保存消息;
3. 使用单独线程检测channel有效性，保证高可用;
4. 所有单小时的数据，保存在内存中，速度特别快；
5. 多线程技术发挥得很好；
6. 模板模式的应用，阻塞队列的应用；
7. hdfs的应用，优雅停机的应用；
等等，来个图展示下：
task 运行过程：
周期报告，汇总：
