百万架构整体流程图2

Link JS SDK(嵌套在推广资料里)

index:summary_hour_behavior_log

公众号收集数据

实时流计算服务

deal-flink-server

deal-weixin-data-server

查询操作

DTS监听PolarDB的binlog

deal-link-data-server

Worker

网站行为收集服务

deal-core2-server

Network Manager

微信公众号

微信事件处理服务

原始筛选条件

规则引擎服务

ES

action

hour

行为数据统计存储

rule_id

通用数据网关服务

无筛选条件查询

开始(点击推广资料)

...

//修改网站访问记录结束时间webAccessLogController.updateLogEndTime()//修改网站访问记录结束时间webAccessLogController.updatePageViewEndTime()//更新网站行为记录结束时间webPageBehaviourLogController.updateWebPageBehaviourLogEndTime()//获取网站访问记录webAccessLogController.getWebAccessLog()//==============创建行为转发到deal-behavior-server服务===============////创建页面访问记录webAccessLogController.before.createWebPageViewLog()->webAccessLogService.createWebPageViewLog()->behaviourLogService.createActionLog()//创建网站行为记录webPageBehaviourLogController.before.createWebPageBehaviourLog()->webPageBehaviourLogService.createWebPageBehaviourLog()->behaviourLogService.createActionLog

gw-wechat-event-kafka-server 

topic_app_page_action_eventtopic_app_page_behavior_eventtopic_app_page_view_event

nginx-server

生产数据

定时任务服务

如果存在查询条件:存储查询条件集合

topic:topic_wechat_events_proceed_00

index:summary_hour_rule_action_log

topic:topic_link_view_log

user用户模型

//创建dealUser行为记录behaviourLogController.before.submitBehaviourDatas()->behaviourLogService.submitBehaviourLogs()->self.createActionLogWithLocation()

deal-robot-server

Memory & I/O Manager

API请求转发

Working Memory（工作内存）

deal_user_id

Agenda(议程)

data-gateway-server

网关服务

kafka

微信网关服务

行为日   志存储、更新

开始(用户访问官网)

nginx

deal-timer-server

返回filter_id

behaviour_log_v3

customer_context客户上下文模型

customer_filter表id:9852c85d-7e61-42fc-bb31-31c6708d2b3cappId:04bb2ceb-fe0a-4866-870a-4a3b36b8e463creatorId:123410createAt:2020-09-23 00:38:15update:

deal-filter-server

gateWay.js//修改网站访问记录结束时间deal-user/:id/web-access-logs/:log_id/end//创建页面访问记录deal-user/:id/web-access-logs/:log_id/web-page-view-logs//修改网站访问记录结束时间deal-user/:id/web-access-logs/:log_id/web-page-view-logs/:view_log_id//创建网站行为记录deal-user/:deal_user_id/web-page-view-logs/:view_log_id/behaviour-logs//更新网站行为记录结束时间deal-user/:deal_user_id/web-page-view-logs/:view_log_id/behaviour-logs/:behaviour_log_id/end

//消费kafka数据并统计每小时dealUser的行为Sink到ESStreamWindowCountForBehavior.main()

deal-customer-search-server

created

行为统一处理服务

TaskSlot

DTS

DTS订阅服务

订阅kafka

行为数   据存储

data-gateway-serevr

链接服务

规则引擎业务模型

微信解密服务

筛选过滤模型关系图

有条件查询

部署图

无条件查询

deal-tag-server

浏览官网收集数据流程

发布&订阅kafka

topic:topic_wechat_events_v3_00

不同类型行为数据入口

JS SDK(嵌套网站) 

Android/IOS SDK

rule规则模型

Dataflow Grap

customer_filter_condition客户筛选过滤条件模型

统计数据SINK到ES

消费行为数据和action数据

接口：GET  /app/{app_id}/screening/customers1、调用筛选条件服务获取筛选条件filter    服务deal-filter-server  2、调用核心core2服务获取元数据metas   服务deal-core2-server3、根据filter和metas整合检索条件，调取ES进行检索

JobManager

行为日   志存储

drools架构图

deal-app-data-server

Android/IOS SDK(嵌套在APP/小程序里)

Actor System

Execution Engine（执行引擎）

开始(公众号交互)

请求筛选条件

Kuick Deal 移动应用端行为数据服务

Inference Engine（推理引擎）

deal-api-server

CheckpointCoordinator

deal_app应用模型

接口：GET  app/{app_id}/customer-metas

customer_meta_data表id:154c3f12-5f4f-4633-8b3d-a73fb42f22f8appId:04bb2ceb-fe0a-4866-870a-4a3b36b8e463name:solomontype:texttitle:solomonisExt:1unique:0required:0visiable:1defaultValue:43createdAt:2020-09-12 18:22:40updatedAt:optionValues:[{\"value\":\"\

deal-web-data2-server

deal-web-data-server

topic:behaviour_log_v3

topic:rule_action_event

deal-weixin-decrypt-server

topic_deal_user_tag

综合筛选服务

customer_filter客户筛选过滤日志模型

行为服务统一处理所有行为数据

deal-server-data-server

//1.消费kafka数据入口ConsumerWechatEvent.receive()->eventProcessor.process()//2.解密微信数据并重新发送到kafkaEventProcessor.process()->switch->producerForOfficialAccount.send()

//发送数据到kafkaevent_controller.lua:producer.send//配置文件lualib/kuick/config_prod.lua

\"query\":{        \"match\":{           \"sc_sex\

rule_action_event

deal-behavior-server

封装筛选条件

deal-customer-search-serevr

count

//消费kafka后的执行流程ConsumerCustomerDomainEvent.receive()->processor.process()(这里选择EventProcessorForEventBehaviorLog实现)->customerDealEngine.customerDeal() {   //1、根据appId从数据库中获取需要执行的规则   //2、规则引擎会编译这个规则，变成可执行的Java代码   //3、规则是根据业务逻辑，自己写的模板脚本(模板参考deal-open-weixin-server服务中resources->velocity          ->rules文件夹下的.vm文件)   //4、使用开源框架drools解析模板脚本            String appId = context.getAppId();               KnowledgeBase kBase = droolKBaseGenerator.getKBase(appId);   kSession.insert(context);  // 执行匹配成功的规则  kSession.fireAllRules();}====================================执行规则流程=======================================1、vm文件中会调用需要执行的action(格式:$ctx.execAction(\"actionName\

topic_link_view_log

JS SDK(嵌套网站)

筛选条件服务

appid

topic_wechat_events_v3_00

访问开课吧产品

2、Pattern Matcher对Fact对象和Rule Base进行匹配

统计每个用户每小时触发规则数

customer_domain_event_v3

消费数据

增、删、改操作

消费行为数据

核心服务

Rule Base（规则库）

customer_meta_data客户元数据模型

customer_id

//1.刷新粉丝数据ConsumerForFansFlush.receive()//2.其他操作ConsumerForOther.receive()font color=\"#ff0000\

deal-api-serevr(请求转发)

第三方平台数据接收服务

summary_hour_behavior_log

客户搜索服务

返回原数据

PolarDB

实时流计算流程图

client

TaskManager

标签服务

gateWay.js//创建网站访问记录deal-user/:id/web-access-logs

API请求转发服务

timer_event_v5

返回筛选条件

ES索引字段

数据网关服务

请求元数据

发送数据到kafka

有筛选条件查询

//消费kafkaViewLogConsumer.ViewLogConsumer()->Mock转发到LinkViewLogController.dealViewLog()//处理日志并发布事件LinkViewLogService.dealViewLog()->publisher.publishEvent(viewLogVo)//LinkViewLogService类中加了@EventListener注解都会监听//entryUrlBehaviour方法监听事件后封装信息发送给deal-behavior-server服务LinkViewLogService.entryUrlBehaviour()->restTemplate.postForObject()

Scheduler

customer客户模型

生成ES查询条件

统计每个小时每个用户触发的每一种规则数

接口：GET  /app/{app_id}/filter/{filter_id}获取指定filterId的条件集合：customerFilterConditionDao.conditionsList(filterId)

customer_filter_condition表id:691ca9ee-c450-4408-b590-13cef17902edfilterId:9852c85d-7e61-42fc-bb31-31c6708d2b3cqueryName:sexqueryType:0queryRange:1createdAt:2020-09-23 00:38:16update:id:78d7aecd-a835-432f-8abd-4212980f8c2efilterId:9852c85d-7e61-42fc-bb31-31c6708d2b3cqueryName:solomonqueryType:0queryRange:123createdAt:2020-09-23 00:38:16update:

生产action数据

统计每个小时每个用户产生各种不同行为的次数是多少统计每个小时每个用户触发的每一种规则数是多少

Data Streams

drools规则引擎执行流程

返回元数据

summary_hour_rule_action_log

rule_action_log规则行为日志模型

接口： GET  /app/{app_id}/member/{kuick_user_id}/customers1、查询所有关联的KuickUserIds   方法：salesCustomerServiceImpl.parseAllToKuickUserIds()2、获取项目开关配置                    方法：salesCustomerServiceImpl.getCustomerSortByUpdateDateConfig()3、获取满足条件的客户                 方法:  customerDao.findCustomersBySales()

Pattern Matcher（匹配器）

customer_filter客户筛选记录模型

推广资料链接收集

dts-subscribe-server 

AccessLogResource.create();参数:       系统内部的deal用户ID:                   dealUserId       网站:                                                site       客户名称:                                         clientName       客户端版本:                                     clientVersion       操作系统:                                         os       操作系统版本:                                  osVersion       网:                                                    net       系统:                                                system       来源网址:                                         referrer       网站访问时的原始地址，包括参数: url

app_member应用成员模型

请求原数据

//发送数据到kafkawechat_event_controller.lua:producer.send//配置文件lualib/kuick/config_prod.lua