平安股票银行_ RocketMQ 在平安银行的实践和应用

随着网络金融业务和相关技术的发展，传统金融业为满足业务快速发展的需要，积极引进各种开源技术，迅速切断市场。 那么，以金融和科学技术为双驱动力的平安银行在开源技术的引进方面是如何评价的，面对怎样的业务场景和复杂的网络环境，是如何发展的呢？

本文以Apache RocketMQ为例，与您一起理解平安银行在开源技术选择中的思路和实践。

RocketMQ在平安银行的应用场景

复杂网络环境下的部署实践

多分隔场景下的部署情况

多IDC方案中的部署

改造实践与邂逅的情节

RocketMQ在平安银行的应用场景

目前，平安银行通过RocketMQ解决数据预加、信息通知和状态变化的业务需求，接着通过App注册、资产概述和工资资产管理三个应用场景开展讲座。

App登录:

当用户打开平安银行App时，系统根据用户的登录ID上载银行卡、信用卡、资产管理产品等用户数据，并接收系统通知。 在这种情况下，预载RocketMQ异步处理能力，即必要的数据，以提高用户体验。

资产概览:

进入平安银行App资产概要页面后，会显示帐户馀额、各种资产管理产品(金、基金、股票等)、贷款等信息。 平安银行的特点是对接来自各种基金、黄金和股票等不同金融主体、不同系统的数据，种类多，不同系统多，变化快。 我们的目标是，资产概要数据尽可能准确，在不同种类的资产变动时发出通知，资产系统收到通知后，根据变化的数据计算用户当前的资产概要。

工资资产管理:

工资资产管理是指每月的工资发行到银行卡后，系统能够自动购入用户设定的资产管理产品，例如购入特定种类的资产管理产品。 这里的信息传递流程如下所示

银行卡馀额波动，系统将波动信息传送至报文引擎

顾客端进行消费，通知用户账号发生了变化

系统会判断变动是否来源于工资

如果适用，将发送另一条消息

资产管理顾问消费

现在判断是否应该购买工资资产管理类的产品

如果是的话，自动购买用户设置的资产管理产品

自动购买后，剩馀金额再次变动，系统再次通知。 这次的通知会判断别的逻辑。

那么，在这些方案中，消息引擎有哪些要求？

a、高可用性、高可靠性、高性能是金融业引入开源技术的基本要求

b、积累能力，支付工资的用户很多，某公司的员工在某个时间点集中支付

c、顺序能力，即账户变动先行，发出通知后

d、没有办公能力、事务性的话，账户可能会发生变化，但没有触发资产管理产品的购买的情况

e，重试和延迟消息功能，例如工资发放时可能会是晚上。 此时，系统在第二天白天操作自动购买资产管理的动作，发出通知

f，消息的回溯能力，在发生问题时，我们可以回溯这个消息，重新处理消息，提供完整的消息回溯

在技术选择过程中，RocketMQ满足了这些典型使用场合对信息产品的需求，在引进过程中，平安银行也为社区做出了贡献。

复杂网络环境下部署的实践

多测试子环境下的服务呼叫场景

平安银行有多个测试子环境，用于测试不同的feature，图中的FAT，FAT001？ FAT002？ FAT003等。 传统的银行系统从大型机时代转向面向互联网用户的零售时代过程中，微服务化是不可避免的。 传统的集中式系统分为许多微服务。 正常情况如下图所示。 服务a调用服务b，服务b调用服务c，服务c调用服务d。

新的功能需要更改服务a和服务b以满足您的业务需求。 比在FAT环境中引入测试，更恰当的方法是FAT环境。 在这里，命名为FAT001，导入服务a和b，a调用b，b调用原来的FAT环境的c和d。

此时，另一项新需求需要改变服务a和c。 直接发布到FAT或FAT001是错误的，会影响正在运行的测试。 在这种情况下，将服务a和服务c命名为FAT002并重建测试环境。 由于FAT002没有服务b，因此服务a必须前往FAT环境以调用服务b ( fat定义为相对稳定的测试子环境)。 当服务b调用服务c时，不应调用本环境中的c，而是激活FAT002中的c以实现测试功能。

并且，如果系统同时另一个feature测试c和d，那么此时的呼叫逻辑也相同，系统呼叫服务a和b时为FAT，呼叫服务c和d时为FAT003的环境。

以上的服务呼叫场景可以通过微服务框架解决，可以进行全链路测试，但在生产环境中，用户的真正要求比测试环境复杂。

实际用户请求进程

让我们看看实际的用户要求。

APP必须启动请求，进入网关，知道请求的测试环境。 测试人员通常需要在APP上选择子环境。 例如，如果选择FAT001，请求FAT001的网关将直接(单独部署到每个环境网关)或向requestheader添加id，以确定网关来自哪个环境(网关集成部署)。 当判断为请求来自FAT002时，所述网关将其分发至FAT002环境中进行处理。

如何在消息级别处理此类用户请求

以上是服务呼叫的要求路径，容易理解，但进入消息队列会使问题变得复杂。 如果feature只是改变了消费者，如何将该信息传递给改变后的消费者进行测试，使其不被其他环境的消费者消费，这是我们应该考虑的问题。

从具体情况来看，集群是Broker和Broke B，TopicA分别配置在这2个Broker上。 此时，Producer Group A将数据写入Topic A，在Consumer Group A消耗的情况下没有任何问题。

但是，在添加了FAT001组的环境下，如果FAT001发出的消息只能消耗FAT001，不能消耗FAT，该怎么解决呢？

是否要将根目录添加到邮件，或者添加Tag、Filter或邮件属性？

这些解决不了我们的问题。 什么东西？

是否在每个子环境中部署RocketMQ？

成本过高，而这个feture测试完成后，相关联的

重新切断FAT进行处理，实现过于复杂。 什么东西？

如果同时测试多个功能，您认为DB会放置一组或多组吗？

首先，feature所属的应用程序不发生变更的原因是，通常配置了DB，将字段添加到数据库中，识别数据来源于哪个子环境，如果有多个配置，则不发生变更的应用程序无法获得新配置的DB数据。 由于不能进行全链路测试，因此同样地，我们也不是在每个子环境中部署RocketMQ，而是部署统一部署，RPC路由将请求路由到正确的生产者集，改造消息路由算法以将消息路由到正确的消费者

实际用户请求进程

生产者变更

在上述图中，假设生产者改变的情景中，默认场景FAT发送、FAT消耗、没有问题地由FAT001的生产者发布，则需要向MQ集群发送FAT001，FAT可以直接消耗。

生产者和消费者同时变更

在上图的生产者和消费者同时变更的情况下，当消费者将应用导入到FAT001中时，FAT消费者不能消费在FAT001中产生的消息，需要消费者消费FAT001。 我们的处理方式通过对Topic A下的Queue进行逻辑分组，相当于增加了逻辑分组，如果消费者配置在FAT001中，则扩大Queue的分组，将其默认设定进一步加倍，新追加的Queue消耗在FAT001中

只有消费者变更

让我们看一下只有消费者才会改变的场景。

假定某个feature仅通过改变消费者就配置在FAT001上。 通过在逻辑上进行分组，生产者可以根据请求源在队列FAT001中向Queue发送消息，从而仅使位于FAT001中的消费者消费。

从以上3个场景可知，需要追加逻辑组，实现过程并不复杂。 主要做了以下调整:

这个逻辑小组什么时候成立？

新设Topic的时候，全部都完成了吗？ 还是在Consumer联机/脱机时动态调整？

您选择了动态创建方法。 在此过程中，我们添加了Meta Server以进行元数据管理，并动态创建了它

添加meta服务。 要管理的元数据包括Producer、Consumer、Topic、Queue和Subenv等信息

调整Producer，若Request Head内的请求源( FAT、FAT001、FAT001 )有对应于topic的组，则选择组的Queue，否则转发至默认组的Queue

调整Consumer，如果不存在联机时应用程序所配置的组( FAT、FAT001、fat002)…topic对应的组，则如果存在新创建的组，则rebalalce (新的Consumer节点联机)、脱机 确定该组中是否存在其他Consumer实例，如果不存在则删除该组，否则rebalalce(Consumer节点脱机)

多分离场景下部署的实践

为重视安全，金融业的网络环境比其他行业复杂。 整个隔离区分为以下几个部分:

DMZ区域:

外联网可以直接访问以配置网关

Web区域:

面向用户的移动电话或可在网页上查看的功能应用程序

核心区域:

与包括核心在内的呼叫逻辑功能、交易和订单相关的核心应用程序(如数据库和存储)

外联区:

您也可以部署Poxy代理程式以存取外部网路。 因为内部网无法直接访问外部网，所以必须通过代理访问外部网

私有区域:

对接基金、三方存款等外部系统。 在金融行业，如果某个系统是闭环，就要隔离

管理区域:

集中管理整个地区的应用程序，数据流是单方面的，只能采集数据，一个地区的数据不能通过管理区域推送到另一个地区。

另外，从安全角度来看，所有地区在生产环境中都通过防火墙隔离，给RocketMQ的部署带来了巨大困难。 只配置一组防火墙，面对如此多的防火墙，从生产者消费者到集群的流量越过墙壁，给网络带来很大的不稳定，成为瓶颈，防火墙几乎在线无法顺利扩展的所有子环境中配置一组

最后，我们采用折中的办法，统一部署加点隔离区的好处如下

防火墙打开大战略，在保证安全的前提下，满足监督管理的要求

对于跨越隔离区消费的问题，我们采用拷贝方式，模拟了消费者重新写入目标集群

在多IDC方案中实施的实践

该城多IDC被认为与LAN相近，比较容易处理，但异地多IDC活跃多。 现在我们没有特别的解决办法。 多元化不可避免地面临数据碎片、数据整合、数据冲突解决等问题。

在Topic中，我们暂时以拷贝的方式处理数据的有效性。 但是，如果以这种手动方式模拟将消费者的消费数据写入新集群的复制方式，则在复制到其他集群后offset发生变化，存在处理逻辑不均匀的问题。 我们根据offset消费用pull方式定制的东西。 如果故障转移消耗到新群集，则必须在新群集中获取正确的offset值。 在这种情况下，该值不同于先前集群的值，因此必须按时间得到新集群的正确offset值并消耗。 在没有更好的解决办法之前，治理至关重要。

但是，RocketMQ的最新版本提供了DLedger的特性，因此DLedger定位了工业级Java库，可以友好地嵌入到任何类型的Java系统中，以满足高可用性、可靠性和一致性的需要。 尽快整合并测试这项功能。

改造实践与邂逅的情节

使用RocketMQ时，添加了以下功能或特性

a .向生产者提供消息积累能力。

b .将所有配置管理部署到配置中心，并云化以满足动态更改的需要。

c .在版本4.3尚未提供事务能力之前，我们将在本地数据库中构建消息表，并在数据库改变数据状态时将数据写入消息表。 如果发送失败，将进行补偿。 在客户端封装。

d .实现统一消息者的应等处理。

e .添加认证和消息认证(注意: RocketMQ 4.3版本实现了认证功能)

当然，也有一些小故事，基本上是使用上的小问题，大家也许会遇到

a .如果应用程序使用多个RocketMQ群集，则未加载到正确的配置中。 在客户端，如果未配置实例名称，则一个应用程序将无法连接到多个群集。

b .在大数据场景中，内存溢出。 订阅的主题越多，每个Queue本地缓存的消息就越多，缺省情况下，每个Queue 1000条内存可能会爆炸，并且可以根据情况进行调整。

c .删除数据时的IO抖动，默认是每天上午4点删除数据，在数量增加后发生IO抖动，构成消息删除策略，用默认的逗号分隔删除数小时即可。

在D. Broker日志报告延迟消息中找不到数据文件。 在备用切换练习期间，延迟消息可能会在Broker中处理异常。 当主要消息切换到待机时，延迟消息保存在Broker中的文件将自动删除，并返回到主要消息，并且存在延迟消息的元数据，因此查询和处理该文件，并且此时找不到该文件。

e .连接nas时，IP取得nas的私人网络地址，提交给服务器端。

以上是我们在开展过程中遇到的一些小故事，基本上很快就能找出原因并加以解决。

综上所述，RocketMQ对平安银行消息系统的建设提供了非常大的帮助，特别是满足了数据持续化、逐步消费和后退的需求，并且在消息查询方面远远优于我们以前使用的消息引擎。 最后分享自己对中间件的意识:中间件的使用管理很重要，规范不先行，开发出两瓶眼泪。