之前讲了 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=' ' 和 lf=' ', 进行分割 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 运行过程：
周期报告，汇总：
