伊始 -- 网聚宝业务监控系统
今年网聚宝开始推进服务化进程，我有幸分在架构组，在公司产品服务化的过程中负责业务监控模块。在以前，我们发展工程还未服务化的时候，我们可能只单单关注单个请求的耗时等基本指标，对其作出优化或者业务上的调整，其内容往往也只是一人从顶到底，一人即可了解其中的逻辑层次。但随着服务化的到来以及业务逻辑愈来愈复杂，一个前端 web 层的请求需要调用很多服务才可以完成一次请求，当请求变的比较慢或者发生故障，我们很难看出是哪个服务或者哪台机器发生的问题，所以需要一个监控服务，来快速复原出完整的请求链，以及请求中的耗时、异常等信息。从而判断是哪里出现了问题，哪里我们还有改进的空间，所以服务化监控模块应运而生。
而我们的监控过程中，部分模块基于了阿里云的基础设施，使得我们在上业务监控的过程中如虎添翼。阿里云的基础设施在使用过程中具有使用非常便捷快速的优势，真正做到了只关注业务逻辑，阿里云也将其旗下各个产品链路打通，我们可以根据自己的需要自由选择组合，完成我们的构想，这种一体化的体验，让我们在上云的时间过程中，真真切切的感受到了云生态。
下面，我将带大家一起来看看，我们在做监控部分中，如何玩转阿里云生态链，构建我们的项目异常监控信息收集。我们使用阿里云数加流计算（Ali Stream）为核心中枢，上下游使用到 阿里云日志服务（SLS） 、 消息队列（ONS） 、阿里云数据库（RDS） 这些产品。SLS + Ali Stream + ONS + RDS 的链路打通也为我们开发过程中带来了极大的便捷。
尤其我们在选择以数加流计算平台作为数据流中枢，数据流的传输过程中我们都不需要写过多的代码，即可完成 上下游数据传输，是不是相当的炫酷呢？下面，我将带领大家一同探究其中的奥秘，选择一些实现细节与大家一同分享，领略下 “无代码编程” 的魅力所在。
构建 -- “无代码编程” 上云
前文提到，我们在针对异常监控的模块中 使用到了 SLS + Ali Stream + ONS + RDS 这条产品链路，那么有看官会好奇，我们是如何使用到这条链路从而达到 “无代码编程” 的呢，在上云过程中我们有遭遇哪些困难呢？
下面是我们业务监控结合阿里云产品链路的架构图：
现在，我们从我们工程中的几个模块来探究这条链路的实现细节：
监控客户端
我们使用在我们的 Dubbo 服务中依赖了网聚宝的自有监控客户端，里面嵌入了诸如调用链监控埋点，异常截获等模块，其主要作用是代码切面拦截等功能，我们对未捕获异常拦截也在此，我们使用 Java 自带的 Thread.UncaughtExceptionHandler 接口 下的 void uncaughtException(Thread t, Throwable ex) 方法对异常进行拦截，并获取异常信息以及异常堆栈。
```java
public class ExceptionLoggerHandler implements Thread.UncaughtExceptionHandler {
private static Logger log = Logger.getLogger(ExceptionLoggerHandler.class);
@Value(value = "${dubbo.application.name}")
private String exceptionDubboName;
public ExceptionLoggerHandler() {
}
public void uncaughtException(Thread t, Throwable ex) {
log.error("Uncaught exception in Application: " + exceptionDubboName, ex);
}
}
```
将异常信息使用阿里云日志服务 com.aliyun.openservices.log.log4j.LoghubAppender 进行收集，并根据 log4j.xml 发送到 相对应的 logstore中去。
log4j.xml 关于SLS部分的配置：
```
<!-- 阿里云监控日志 -->
<appender name="loghub" class="com.aliyun.openservices.log.log4j.LoghubAppender">
<!--阿里云日志服务信息配置-->
<param name="projectName" value="${projectName}" />
<param name="logstore" value="${logstore}" />
<param name="endpoint" value="${endpoint}" />
<param name="accessKeyId" value="${accessKeyId}"/>
<param name="accessKey" value="${accessKey}"/>
<!-- 填写监控的dubbo application名称 -->
<param name="topic" value="${your dubbo application name}"/>
<param name="packageTimeoutInMS" value="3000"/>
<param name="logsCountPerPackage" value="4096"/>
<param name="logsBytesPerPackage" value="5242880"/>
<param name="memPoolSizeInByte" value="1048576000"/>
<param name="ioThreadsCount" value="1"/>
<param name="timeFormat" value="yyyy-MM-dd HH:mm:ss"/>
<param name="timeZone" value="GMT+8"/>
</appender>
<logger name="loghub" additivity="false">
<level value="ERROR" />
<appender-ref ref="loghub" />
</logger>
<root>
<level value="WARN" />
<appender-ref ref="loghub" />
</root>
```
根据上面的配置，当项目中发生异常时，就可以在日志服务中就可以收到日志信息了。
到这里数据收集的工作就算完成了，下一站，Ali Stream！
阿里云流计算配置
现在我们数据到达了流计算这个中枢神经中，在 流计算中 我们在开发平台构建源表，数据源当然是 日志服务啦。
SLS 源表：
```
CREATE STREAM TABLE monitor_client_log_input (
message STRING,
time STRING ,
level STRING,
`location` STRING,
thread STRING
) WITH (
type='loghub',
endpoint='${endpoint}',
roleArn='${roleArn}',
projectName='${projectName}',
logstore='${logstore}',
allowKeyNotExists=TRUE
);
```
流计算 “数据中枢” 的工作目前非常简单，收集到数据，重组数据，一边通知 ONS ，一边存储到 RDS 中，所以为了达成这个目的，我们需要构建两张结果表，分别指向不同的地址。
根据需要通知监控服务端的内容建立 ONS 结果表：
```
CREATE RESULT TABLE monitor_server_log_collect_exception_ons_result (
message STRING,
time STRING ,
level STRING,
`location` STRING,
thread STRING,
`topic` STRING
) WITH (
type='mq',
endpoint='${endpoint}',
topic='${topic}',
producerId='${producerId}',
accessId='${accessId}',
accessKey='${accessKey}',
fieldDelimiter=",",
domainSubGroup="nsaddr4client-internet"
);
```
根据RDS的对应列建立 RDS 结果表：
```
CREATE RESULT TABLE wjb_application_exceptions_rds_result
(
`error_application_name` string,
`error_stack` STRING,
`gmt_create` timestamp,
`status` int
) with (
type = 'rds',
url = '${url}',
username = '${username}',
password = '${password}',
tableName = '${tableName}',
cacheTTL = '120000',
cacheType = 'LRU'
);
```
OK ，表的构建就结束了，下面我们当数据到来的时候，我们需要将数据插入进表格就可以了。
```
--通知 ONS
INSERT INTO TABLE monitor_server_log_collect_exception_ons_result
SELECT
`message` as message,
`time` as time,
`level` as level,
`location` as `location`,
`thread` as `thread`,
prop_get('__topic__') as `topic`
FROM monitor_client_log_input
WHERE
level = 'ERROR';
--插入 rds 异常表
INSERT INTO TABLE wjb_application_exceptions_rds_result
SELECT
prop_get('__topic__') as error_application_name,
`message` as error_stack,
`time` as gmt_create,
1 as `status`
FROM monitor_client_log_input
WHERE
level = 'ERROR';
```
TIPS：
我们在 日志服务 中，使用 topic 作为标识符 标示不同的异常服务，所以这个是loghub的系统字段 topic, 在 获取数据时 使用 prop_get('__topic__') 获得字段值。使用系统参数，还需要配置下任务参数（右边栏）。
在任务参数中添加：
`galaxy.input.props.keylist=string,__topic__`
好了，到此为止，这个 “数据中枢” 就构建完了，只需要点击 上线，启动，就OK了，是不是 so easy ？
启动之后可以在运维模块下 看到当前 流计算的状态：
监控服务端
服务端的工作只有一件事：
通过 建立 ONS 消费者监听，实时获得 异常信息，通知前端，以及响应及时处理机制。这样，我们就可以比客户反馈，快一步去处理去处理这些异常。
```java
Properties properties = new Properties();
properties.put(PropertyKeyConst.ConsumerId, MonitorConstants.ONS_CID_EXCEPTION_HANDLER);
properties.put(PropertyKeyConst.AccessKey, MonitorConstants.ACCESS_KEY_ID);
properties.put(PropertyKeyConst.SecretKey,MonitorConstants.ACCESS_KEY_SECRET);
Consumer consumer = ONSFactory.createConsumer(properties);
consumer.subscribe(MonitorConstants.ONS_TOPIC_MONITOR, "*", (message, context) -> {
//启动异常处理机制
return Action.CommitMessage;
});
consumer.start();
```
成果 -- 先客户一步
到这里，异常监控的一条 基于阿里云的 “无代码编程” 链路就打通了，在我们的开发过程中，我们基本不需要写过多的代码，就可以将我们需要做的事情融入到 网聚宝的系统中，我们只需要真正关注我们的业务逻辑，我们的 异常监控数据 “腾云驾雾”，先客户一步到达我们的眼前，当客户来因为异常问题找到我们的时候，我们已经找到问题，处理完毕了，减少了交流的成本，让客户使用我们产品体验也得到巨大的提升。
网聚宝服务化的进程还在推进，我们云上的架构也在逐步完善扩充，根据阿里云生态的体系，我们相信可以找到更多适用于我们自己的“最佳实践“。
上海云贝网络科技有限公司
架构师 童煜