2GIS的用户每天都在帮助我们保持数据的准确性：他们为新公司提供信息，增加交通事件，上传照片并撰写评论。 以前，我们只能用言语感谢他们或安排赠品。 但是随着时间的流逝，单词被遗忘了，并不是每个人都能得到礼物。 因此，我们决定确保所有关心2GIS的人都能看到他们对产品的贡献以及对此的感谢。
因此获得了奖项-我们为各种任务获得的虚拟勋章：将照片上传到咖啡馆卡，撰写有关剧院的评论，指定组织的工作时间等等。 用户可以在自己的2GIS个人资料以及移动应用程序的“我的2GIS”选项卡上看到所获得的奖励。 在那里，我们显示了下一个成就还剩下多少。
为了实现此功能，我们学习了如何处理事件流，每小时事件处理量为50万条记录（每秒最多可处理5万条记录），并分析来自多个服务的数据。 而且-他们增加了一些元编程，以便在获得新奖项时简化配置。
与
Rapter一起，我们
将告诉您奖励过程的内容。
概念图
为了了解功能的复杂性，您需要了解技术问题的发音。 然后-考虑实现的思想和系统组件的一般方案。 这就是我们在本节中要做的。
摘要要求
需求-相当无聊的事情，因此我们将不会画出所有细微差别，我们将专注于最重要的事情：
- 奖励仅发给授权用户；
- 更新奖励进度应尽可能快；
- 奖励-用户在产品中执行一系列操作的结果：上载照片，撰写评论，查找路线等。数据来源很多。
建筑理念
实现的想法不是很复杂。 可以用论文来表达：
- 奖励由任务组成，任务的结果根据配置奖励时指定的公式进行组合；
- 该任务响应来自外部的用户操作事件，对其进行过滤并以计数器形式记录进行中的更改；
- “外部事件”是由主系统（照片，反馈，优化服务等）或辅助服务生成的，这些主服务会转换或过滤现有事件流；
- 事件处理是异步发生的，必要时可以随时停止；
- 用户看到其奖励的当前状态；
- 其他一切都是细节...
关键实体
下图显示了主题区域的主要实体及其关系：
图中区分了两个区域：
- 计划-描述奖励结构和应计规则的区域；
- 数据-特定用户的奖励区域以及与他们当前状态有关的数据。
图中的实体：
- 达成-可获得的有关奖励的信息。 包括元信息和有关如何组合任务结果的描述-一种策略。
- 目标-一项任务，必须满足其条件才能前进到获得奖励。
- UserAchieve-特定用户的奖励的当前状态。
- UserObjective-用户奖励工作的当前状态。
- 用户-有关用户的信息，对于进行通知和了解其当前状态是必需的（不需要远程和禁止的奖励）。
- ProcessingLog-任务应计的日志。 包含有关特定操作如何影响分配进度的信息。
- 事件-关于某种事件的最小必需信息，该事件以某种方式影响了用户任务的进度。
服务架构
现在考虑服务的主要组件及其依赖性：
- 事件总线-可用于完成任务的事件总线。 我们正在使用Apache Kafka。
- 主数据库和从数据库是主要的数据仓库。 在这种情况下，将使用PostgreSQL集群。
- ConsumingWorkers-总线事件处理程序。 主要任务是从特定来源（照片，评论等）读取事件，将其应用于用户任务并保存结果。
- AchievesWorker-根据任务状态重新计算用户奖励的进度。
- NotificationWorkers-一组处理程序，用于安排和发送有关接收奖励，新可能成就的公告等的通知。
- 公共API-用于Web和移动应用程序的公共REST接口。
- 专用API-用于管理面板的REST接口，可帮助进行事件调查和服务支持。 开发人员和支持团队可以使用它。
每个组件在逻辑和职责范围方面都是隔离的，从而避免了在修改数据时不必要的集成和死锁。 下面我们仅考虑与事件处理并将其转换为奖励相关的部分方案。
事件处理
内容内容
奖励主要是数据聚合服务。 每个主系统都会生成几种类型的事件。 通常，每种类型的事件都与内容的状态及其状态模型密切相关。 因此，可以调节，删除，阻止，隐藏或激活照片。 所有这些都是由专门研究特定来源的独立工人处理的不同事件。 目前，与以下来源（主系统）存在交互：
- 照片-生成与用户对照片执行的操作有关的各种事件。
- 评论-与用户评论操作相关的事件。
- 数据反馈-与优化操作相关的事件。 澄清是指关于地图上某个对象（无论是公司还是纪念碑）的信息的更改。
- 检查-与2GIS Check应用程序相关的事件。
- BSS是生成2GIS应用程序的分析事件。 例如，某家公司的开业，在导航仪上旅行等。
由主系统生成的事件按照其状态更改的顺序落入Kafka主题，这使得不仅可以向前移动用户，还可以向后移动奖励的进度。 例如，如果照片处于“活动”状态，然后由于某种原因获得了“已阻止”状态，则奖励的进度应向下改变。 奖励进度是对称为内容计数器的内部对象的解释。
对于不同的数据，计数器可能会有所不同。 例如，对于有关照片的事件，它们如下：批准的数量，节制的数量，被阻止的数量以及对于打开卡片的事件，您只需要考虑用户打开的卡片数量。 根据内容计数器的当前值，对于特定用户，在特定奖励的框架内，确定以下问题的答案：
过滤器和规则
仅当具有所需内容类型的事件以及获得奖项所需的必要数据到达时，特定奖项的工作计数器才会更改。
为了仅跳过适合获奖的内容，我们通过一系列过滤器和规则来运行每个事件。
过滤器是对内容施加的一定限制。 他只关心回答以下问题：“新事件是否符合此条件？”
规则是一个特殊的过滤器，其目的是说：“如果一个事件符合条件，那么计数器应该如何变化？” 该规则包括用于更改计数器的算法。 每个奖项仅包含一个规则。
过滤器和规则的实现在项目代码中，而属于特定奖项的过滤器（规则）的描述在JSON格式的数据库中。 我们没有立即做出这样的决定。 最初，无法使用数据库中的配置来设置过滤器和规则，奖励在代码中得到了充分描述，只有其标识符存储在表中。 该决定带来了许多重大缺陷：
- 支持多种环境的问题。 如果要向测试环境推出奖励列表的一种状态，然后将另一种状态发送给战场，则需要知道项目代码中的环境代码或具有包含奖励列表的配置文件。 同时，尽管对于每个环境已经存在不同的数据库，但是无法使用不同的数据库来完成此任务。
- 只能由开发人员配置过滤功能。 由于所有内容均在代码中进行了描述，因此只有知道项目和编程语言的人才能进行更改，我希望可以仅通过Private API或数据库来进行更改。
- 查看的缺点。 有很多奖励，有时您需要查看它们使用的过滤器。 每次，通过查看代码来做到这一点都是很乏味的。
在应用程序开始时，我们根据从数据库加载的过滤器的名称进行匹配，并将其放入特定的奖励中。 过滤器描述示例：
[ { "name":"SourceFilter", "config":{ "sources":["reviews"] } }, { "name": "ReviewsLengthFilter", "config": { "allowed_length": 100 } } ]
在这种情况下，我们将仅接受那些评论（由过滤器数组中的第一个描述对象指示），其文本包含100个以上的字符（列表中的第二个过滤器）。
规则说明示例：
{"name": "ReviewUniqueByObjectRule","config":{}}
仅当用户为该对象撰写评论时，此规则才允许您更改计数器，而一个对象仅考虑一个评论。
s
让我们分别处理BSS事件流。 至少有以下三个原因：
- Analytics事件无法回滚，其中没有状态模型，这通常是合乎逻辑的，因为无法取消通过导航器的驾驶或构建路线。 行动是否在那里。
- 卷。 让我提醒您，2GIS的总受众每月为50+百万用户。 他们一起进行了超过15亿次搜索查询，以及许多其他操作：启动应用程序，查看对象的卡等。在高峰期，事件数量可以达到每秒50,000。 我们必须通过过滤器传递所有这些信息，以奖励用户。
- Analytics事件具有以下特征：几种格式，多种类型。
所有这些都极大地影响了BSS主题中数据的处理，因为如果我们需要实时，那么就需要非常短的处理时间。
为了减少所描述的差异，已创建了准备此类事件的单独服务。 该服务可以处理来自分析的各种消息格式。 他的工作实质如下：读取整个BSS事件流，从中仅获取颁奖所需的事件。 这样的服务过滤器显着降低了来自BSS流处理器奖励的负载（过滤后，流量约为每秒300个事件），并且还以单一格式生成事件，从而消除了与内部分析历史相关的缺点。
获奖情况
因此，我们弄清楚了如何处理事件并计算作业进度。 现在是时候看看向用户发放奖励的过程了。
出现的第一个问题是：为什么将输出分配给单独的工作程序，在处理每个事件时不能将其重新计数？ 答：可能，但不值得。
将引渡分配给一个单独的进程有多个原因：
- 将重新计票转移到每个ConsumingWorker，我们就获得了通过奖励更新进度的操作的竞争条件，因为每个处理程序将尝试根据任务的已知状态更新进度，而其他处理程序将主动更改此状态。
- 每个ConsumingWorker批处理事务中来自Kafka的事件。 通过在用户的奖励表中添加插入内容，我们将在数据库级别调用额外的锁，这将禁止其他处理程序。
- 在发放奖励的过程中，存在发送通知的逻辑，这只会减慢事件流的处理速度，这是不希望的。
找出了单独的AchievesWorker（颁发奖励的处理程序）出现的原因。 现在，您需要处理处理的两个重要部分：
- 奖励中有一系列任务。 这些任务有一组计数器。 如何理解该奖项的金额以及如何用代码表达？
示例：您需要写3条评论或上传3张照片。 用户有1条评论和2张照片。 奖项进展如何？ 答案：3，因为用户肯定会确定您总共需要3个。 - 我们有一个单独的处理程序来颁发奖励。 每次为每个授权用户重新计算几十个奖项，即几千万个奖项，都不太可能很快成功。 他如何了解自上次处理以来哪些特定用户的进度以及哪些任务已更改？
我们将分别考虑每个部分。
进度流
为了更好地理解您如何描述如何通过奖励将任务的进度转变为进度，我们将奖励划分为几类，并查看转换。
“每X个单位完成一项任务。” 示例：在导航仪上行驶10公里。
“为X个单位分别完成几个任务。” 示例：上传5张照片，并在卡片中写5条评论-内容只有10个单位。
“总共要为X个单位完成几个任务。” 例如：撰写5条评论或上传5张照片。
“完成按类型分组的几个任务。” 示例：上传5个单位的内容（照片或评论），并在导航器上行驶10公里。
从理论上讲，可能会有更复杂的嵌套组合。 但是，在实际条件下，不可能用两句话或三句话向用户解释为获得奖励而必须执行的复杂逻辑组合。 因此，在大多数情况下，这些选项就足够了。
我们将转换方法称为一种策略，并试图通过以JSON对象的形式制定正式的说明来使其或多或少具有通用性。 当然，您可以考虑以公式的形式进行编写，但是随后您将不得不使用eval的相似性或描述语法并实现它，这显然是过于复杂了。 将每个奖项的策略存储在源代码中不是很方便，因为奖项的描述（数据库中的一部分，代码中的一部分）会撕裂，并且将来也将不允许未经开发人员参与而从现成的组件中收集奖项。
该策略以树的形式表示，其中每个节点：
- 指代当前的分配进度，或者是一组其他节点。
- 可能有一个最高限制-实际上表明需要使用min（）。
- 可能具有归一化系数。 通过将结果乘以数字来进行简单转换。 我们派上用场了。
为了描述以上示例，一个操作就足够了-总和。 Sum非常适合用一个数字清楚地显示用户进度，但是如果需要，可以使用其他操作。
这是最后一个类别的示例策略描述：
{ "goal": 15, "operation": "sum", "strategy": [ { "goal": 5, "operation": "sum", "strategy": [ { "objective_id": "photo" }, { "objective_id": "reviews" } ] }, { "goal": 10, "operation": "sum", "strategy": [ { "objective_id": "navi", "normalization_factor": 0.001 } ] } ] }
必需的更新
有多个处理程序可以不间断地分析用户的事件，并将更改应用于任务的进度。 定期搜索每个奖项的所有用户将导致对数千万个奖项的分析-如果真正的更新以成千上万个为单位，这不是很令人鼓舞。 如何只学习数千个而不浪费数百万个CPU？
关于如何仅根据实际上已经改变的奖项重新计算进度的想法很快就出现了。 它基于矢量手表的使用。
在描述之前，我将提醒实体：
- UserObjective-通过设置奖励来获得有关用户进度的数据。
- UserAchieve-奖励用户进度数据。
实现看起来像这样：
- 我们在PostgreSQL中获得了UserObjective和UserAchieve和Sequence的版本字段。
- UserObjective实体的每次更新都会更改其版本。 该值取自序列（我们对所有记录都具有此值）。
- UserAchieve的版本值将确定为关联的UserObjective的最大版本。
- 在每个处理周期，AchievesWorker都会搜索没有UserAchieve或UserAchieve.version <UserObjective.version的UserObjective。 通过对数据库的单个查询即可解决该问题。
立即值得注意的是，该解决方案在奖励和任务表中的条目数量以及任务进展的变化频率上都有局限性，但是有了数千万个奖励并且每分钟更新的次数少于一千，这种解决方案很可能适用。 我们将以某种方式分别说明如何优化竞赛“
2GIS Agents ”的发行。
结论
尽管事实证明这篇文章篇幅如此之大，但幕后仍有许多细微差别，因为不可能简短地谈论它们。
凭借这些奖项，我们得出了哪些结论：
- 在这种情况下，“分而治之”的原则就掌握了。 将事件处理程序分配给每个源有助于在必要时进行扩展。 他们的工作根据数据是孤立的，并且仅在小范围内相交。 突出显示奖励逻辑使您可以减少事件处理程序中的开销。
- 如果您需要消化大量数据并且处理成本很高，则应立即考虑如何过滤掉绝对不需要的内容。 过滤BSS流的经验就是一个例子。
- 我们再次确信，通过公共事件总线进行的服务集成非常方便，并且可以避免不必要的其他服务负载。 如果“奖励”服务通过http-requests从“照片”，“评论”等服务接收了数据，则必须准备好几个服务以增加负载。
- 一点元编程可以帮助维护数据配置的完整性和任意分离的环境。 在数据库中存储过滤器，规则和策略可简化开发和发布新奖项的过程。