本文选自《交易技术前沿》总第三十八期文章（2020年3月）
肖钢、赵眸、聂超 /中信建投证券信息技术部
邮箱：[email protected]
摘要：伴随资本市场对外开放的步骤加快，外资券商也已正式进入中国，整个行业竞争加剧，为增强证券公司竞争力，增强用户体验，客户信息通知与服务的MOT系统显的尤为重要，本文重点讲述中信建投证券MOT系统搭建的实践以及未来规划等相关内容。
1.概述
客户在证券公司开户并进行各种投资操作，在不同的时间点会产生大量的信息需要提醒，每个需要提醒客户或员工的时刻，我们称之为一个MOT点，例如在客户开户的多个步骤中若长时间停留在某一步骤，则需要提醒员工去帮助客户开户；客户开户成功后可以给其推荐新客理财产品；客户新股中签后需要提醒客户已中签等。按照MOT点的提醒时效性，我们将其又分为非实时MOT及实时MOT，例如提醒客户其持仓股票出现了快速拉升，提醒的时效性很高属于实时MOT，而祝福客户生日快乐的短信则时效性较低，属于非实时MOT。为满足不同时效的MOT提醒需求、赋能业务人员在不同的MOT时刻去为客户服务及营销，本文结合中信建投证券MOT系统的搭建过程，探讨如何设计稳定高效的全渠道MOT通知系统。
2.MOT系统实现
2.1系统架构
中信建投证券的MOT系统主体架构如图1所示，从上游各数据源获取数据，加工信息后推送至下游的全部通知渠道，共包括数据处理层、数据层、业务处理层三部分。
数据源部分为MOT系统可接入的数据源，可接入的数据源分为三大类：Kafka实时数据源、EverQuotes实时行情数据源以及各关系型数据库数据源。其中Kafka实时数据源以及EverQuotes行情数据源均为实时数据处理引擎提供数据，需要做实时MOT提醒的数据，均需要交易系统等上游关系型数据库实时同步至Kafka；各关系型数据库数据源主要为非实时数据处理引擎提供数据，数据库类型包括Oracle、SqlServer、Mysql等各类关系型数据库，在数据源的选取上，尽量从数据仓库获取数据，减少因为各交易系统数据结构变动导致的MOT系统被动修改，同时尽量选取上游做过整合的稳定资讯数据源，避免因个别资讯提供商提供数据不准确导致MOT对客户发送错误提醒。
数据处理层包括实时数据处理引擎及非实时批处理引擎，是整个MOT系统的核心计算模块。实时数据处理引擎处理Kafka中获取的实时数据以及EverQuotes中的实时行情数据，可以采用Apache Storm流数据处理引擎，也可采用Apache Spark Streaming处理；非实时批处理引擎定时从上游各关系型数据库中抽取数据并按MOT规则计算。数据处理层加工出各MOT点待发送的消息内容。
数据层为MOT系统的数据存储，包括内存缓存以及硬盘持久化存储，采用Redis数据缓存、Kafka数据缓存以及Oracle持久化存储，同时，采用数据库延迟写技术的数据都会直接在日志中存储，避免数据丢失。我们在部署时将Redis、Kafka及Oracle都做了集群部署，单台故障对系统没有影响。
业务处理层为数据处理层提供基础配置和缓存数据支持，对数据处理层的结果做后续处理，并对外提供查询。业务部门可以在此对各MOT点的参数进行配置，比较灵活的配置MOT点发送渠道话术以及计算规则，查询历史发送情况。后台管理员也可以在业务处理层做任务配置，增加新的MOT点，以及检查各模块运行情况。
图1 MOT结构图
2.2数据处理层主体模块
2.2.1 实时数据处理引擎
Apache Storm、Apache Streaming、Flink等都是较为成熟的流处理框架，Apache Storm是一种侧重于极低延迟的流处理框架，具备如下优点：
I. 数据处理时延小。消息到来之后即时处理，不是积攒到一批后统一处理；
II.具备较好的可靠性保障。处理出错的消息会被再次处理，保障重要的消息通知不会因错误被丢弃；
III．框架支持热发布。新增的消息可热发布到服务器，不会影响生产环境正在执行的其它消息类型；
实时数据处理重在计算，其数据量相对较小，存储不需要使用Hadoop等大数据存储。经过压力测试，三台集群部署的Storm服务器，配置六个Worker，可以及时处理全市场的证券实时行情、根据行情加工出MOT事件（约1200条/秒），不会导致行情处理出现积压，同时还能处理其它交易类实时MOT消息（约5000条/秒）。因此我们最终采用Storm作为实时数据处理引擎。Storm实时消费Kafka中的信息，以Bolt为基本处理单元对Kafka消息进行处理，每个Kafka消息由多个Bolt串行处理，每个Bolt都对消息做一层简单加工处理，加工过程中需要与Redis缓存多次交互，从中获取MOT点配置信息，以及客户持仓、客户经理挂接关系、客户基本信息等基础数据，同时，将需要存储到文件系统及数据库的数据写入Redis缓存，由缓存数据维护模块将数据维护进数据库及文件系统。Bolt加工出最终需对客户及客户经理提醒的信息内容，写入待发送消息队列缓存，由消息发送引擎处理并发送。Storm的处理结构如图2所示。
图2 实时数据处理结构图
为提高系统处理效率，Bolt中的处理逻辑仅与缓存交互，全内存操作，不直接读写硬盘，且将数据流量大的Topic消息在Kafka中做分区存储，提高消息处理的并行度。例如每日在开市集合竞价和闭市集合竞价期间都会产生大量的客户委托失败消息，我们将此消息在Kafka中分成四个分区，开四个线程同时处理可以确保提醒消息在3秒内推送出去（含调用下游通道发送时间）。同时，为确保系统稳定，在部署时需将Storm集群多机部署，处理单元异常后自动迁移至其它服务器。
2.2.2非实时批处理引擎
非实时处理引擎根据任务配置表定时执行任务，从包括交易系统在内的各上游数据库抽取数据并按非实时MOT点规则进行计算，计算之后将加工的信息存放进待发送消息队列，由消息发送引擎处理并发送。非实时批处理引擎的数据处理流程如图3所示。
图3 非实时批处理流程图
在设计非实时任务时，优先导入基础数据，加工出基础数据指标表，再同时并行多个任务计算非实时MOT点。各事件计算均通过中间表存储，计算完毕写待发送消息队列及数据库，避免多个任务同时操作数据库某张表造成死锁。
2.3业务处理层主体模块
业务处理模块为实时数据处理引擎及非实时批处理引擎提供服务，其中XDATA管理模块管理ETL数据抽取配置，按照设定时间判定上游数据是否已准备好并执行对应的ETL配置，将数据从上游各数据库抽取到MOT本地数据库，设计时尽可能的增量获取数据以减轻数据库压力。
消息发送引擎从待发送消息队列获取数据，根据消息类型调用不同的API接口发送，将信息推送至客户短信、手机APP渠道、微信公众号、通达信、同花顺、智能语音提醒等渠道，对员工支持微信企业号、邮件、CRM等渠道。对于未发送的消息优先调用各下游渠道的批量API接口，并采用多线程处理以提高效率，消息发送引擎支持分布式部署，在发送压力较大时可增加部署扩大发送能力。
缓存数据维护模块负责更新Redis中的缓存数据，每日在XDATA管理模块采集完上游数据之后将客户持仓、客户基本信息、客户与员工挂接关系等基础数据更新至缓存，同时，实时数据处理引擎将需要写硬盘的数据都放在redis缓存，例如从Kafka中消费的源消息内容、给客户的短信发送内容、给员工推送的企业微信内容等，缓存数据维护模块负责将这部分数据保存至硬盘做持久化存储。系统支持界面操作手工刷新某一类型数据缓存，手工更新的数据可通过界面操作刷新至缓存；
事件配置模块配置各MOT事件的详细参数，包括MOT点的运行时间段，员工自定义的参数、事件触达到员工及客户的发送模板等，配置的事件信息将同步更新缓存。实时数据处理引擎及非实时批处理引擎依据此配置计算各MOT点事件，事件配置界面如图4所示。每个MOT点支持员工做参数个性化配置以及发送模板配置，其配置的参数仅对员工本人名下的客户有效，如图5及图6所示。
图4 实时MOT点配置图
图5 实时MOT点参数配置图
图6 发送模板配置图
权限控制模块管理用户操作MOT系统的权限，部分特别重要的MOT点例如客户强制平仓，均由开放权限的业务部门手工验证和发送；报表模块提供列表查询及图表查询功能，方便员工查询到自己名下客户曾收到的信息，分支机构可查询到本部门客户的信息送达情况。
3.系统应用
目前中信建投实时与非实时MOT系统在实现此架构的基础上共运行MOT点380个，涵盖普通账户业务、两融业务、股票质押、行权融资等各类业务，日均给客户及员工推送信息约五十万条，实时数据处理从触发到送达的平均时延在秒级别，初步验证了架构的有效性与实用性。
为确保信息推送的准确性及实时性，尽可能降低发送错误，MOT系统配置了一系列的运营保障措施：各MOT点单独设置发送开关，试运营阶段关闭开关，验证无误后再打开开关放开发送；每个MOT点单独设置发送周期，在特定时间内不给同一发送对象发送相同内容；对于超过设定时限的实时MOT点信息不再发送；同时，对架构设定一系列的监控措施，日志中出现异常、ZooKeeper、Redis出现异常、kafka消费慢及时延高时及时报警，以便快速处理错误。另外，从两个方面推进系统建设可有效降低MOT系统报错的概率：一是推进上游建设统一的资讯中心，由资讯中心对各资讯数据源数据比对，确保资讯数据质量；二是推进上游数据仓库的完善，避免因直接对接各交易系统导致的频繁修改。
4.结论与展望
MOT作为客户通知与客户服务工作的重要工具，主要需要做到配置灵活、触达客户及员工准确及时、运行稳定。业务部门可以灵活调整事件计算参数，配置发送渠道及发送话术，在对客户服务的同时也能进行营销，不仅提升客户满意度，还能在客户服务的同时有针对性的推荐产品。后续我们将与业务部门继续合作扩大MOT点覆盖范围，同时对已经运行的MOT点加强运营分析，分析各MOT点运行效果，并以运行效果数据为依据对现有的MOT点进行调整，将MOT赋能业务部门的效果最大化。
