大家好，我是大胜。最近大胜在工作中遇到这样一个数据开发需求，就是监控用户在某个产品的哪些界面上发生的行为，其实简单来说就是计算用户在我们产品的一些界面中的PV/UV，以及用户上一次操作界面和下一次操作界面之间的行为数据。我的想法是和前端人员约定好这些接口的事件点击类型，然后使用Flink框架从Kafka实时消费数据，为每个事件点击类型写一个Flink作业，然后在具体的Flinkjob实现行为指标来监控用户。思路如下图：这个方案确实可以，我也是这样实现的。最后写了8个Flink作业。但是在写代码的过程中发现了很多重复的代码，因为8个Flink作业可能分组字段不同，或者每个Flink作业的处理细节不同。但是如果我这样写那么多作业，维护起来会很麻烦，而且代码的扩展性不好，而且我的每个Flink作业都会从Kafka中消费同一条数据，所以这是不可取的，所以这里可以使用工厂设计模式，将多个Flink作业合并为一个Flink作业。本文主要讲的是使用工厂设计模式将多个Flink作业组合成一个Flink作业，所以工厂设计模式这一部分主要介绍最简单的工厂设计模式，如下图所示：这种模式是最简单的工厂设计图案。因为它实际上封装了不同类型对象的创建，通过将不同的点击事件类型传递给工厂模式来指定要创建的对象。这是本文中使用的方法。下面主要讲一下如何在FlinkJob中使用工厂模式合并多个FlinkJob。话不多说，直接看上图：这是我们从Kafka消费数据后，将数据封装到不同点击事件类型的Bean对象中。我这里创建了一个BaseBean，然后为每一种点击事件创建自己的Bean对象，让每一种点击事件都继承BaseBean的父类。然后使用工厂模式根据不同的点击事件类型创建不同的Bean对象，如下图所示：这是使用工厂设计模式根据不同的用户行为创建类中initBean()方法中的初始化不同的点击事件类型我们自己的Bean对象，如下图所示：这样我们就可以在流中返回BaseBean，使用这个BaseBean继续往下游传递。这个地方利用了Java多态中的向上转化，多个子类转化为一个父类。我们来看看FlinkJob中流程的框架代码，如下图所示：当BaseBean到达process方法时，我们此时会实现计算PV/UV的逻辑。这时候，我们仍然可以使用点击事件类型，将不同的点击类型，交给不同的行为类进行处理，如下图所示：我们将计算结果封装到行为类中的一个BaesModel，让每个behavior类也创建一个计算结果的实体类来继承BaseModel，这里和上面是一样的，然后我们把BaseModel下游传给addSink算子，如下图：最后存入数据库的时候，我们还是使用不同类型的点击事件来执行不同的行为代码，但是因为它是BaseModel是一个父类，所以我们需要将它转化为各个具体的子类。这个地方需要用到Java多态中的向下改造，如下图：这样可以使用工厂设计模式实现多个FlinkJobMerge成一个FlinkJob，而且程序扩展性更好，如果有的话是什么新的业务指标，我们可以在每个行为类中进行扩展。如果文中有不清楚的地方，或者想要文章的示例代码，可以扫描下图，关注微信公众号点击联系我，加我微信与我交流。本文由博客多发平台OpenWrite发布！
