消息订阅模板修改

ldt-common/pom.xml

@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>GUOXIN_LDT</artifactId>
+        <groupId>org.springblade</groupId>
+        <version>2.8.2.RELEASE</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>ldt-common</artifactId>
+
+
+    <dependencies>
+        <!--  糊涂工具包    -->
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+            <version>5.7.15</version>
+        </dependency>
+        <!--  lombok   -->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.20</version>
+        </dependency>
+        <!--  alibaba fastjson -->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.41</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springblade</groupId>
+            <artifactId>blade-starter-log</artifactId>
+        </dependency>
+    </dependencies>
+</project>

ldt-common/src/main/java/com/ldt/consts/QueueConst.java

@@ -0,0 +1,22 @@
+package com.ldt.consts;
+
+/**
+ * @author: lianghanqiang
+ * @description: 队列名字
+ * @since: 10/28/21 -- 5:35 PM
+ */
+public interface QueueConst {
+	/**
+	 * 	api log 接口请求日志输出队列
+	 * */
+	String apiLog =  "API_LOG_QUEUE";
+	/**
+	 * 	err log 错误日志输出队列
+	 * */
+	String errorLog =  "ERROR_LOG_QUEUE";
+
+	/**
+	 * 	IP 高频访问告警 输出队列
+	 * */
+	String ipAlarm = "IP_ALARM_ALERT";
+}

ldt-common/src/main/java/com/ldt/entity/ApiLogData.java

@@ -0,0 +1,36 @@
+package com.ldt.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Date;
+
+/**
+ * @author: lianghanqiang
+ * @description: api 传输数据
+ * @since: 10/27/21 -- 5:51 PM
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ApiLogData {
+	String tenantId;
+	String remoteIp;
+	String env;
+	String title;
+	String type;
+	String time;
+	String serviceId;
+	String serverIp;
+	String serverHost;
+	String userAgent;
+	String requestUri;
+	String method;
+	String methodClass;
+	String methodName;
+	String params;
+	Date createTime;
+}

ldt-common/src/main/java/com/ldt/entity/alarm/IPAlarm.java

@@ -0,0 +1,21 @@
+package com.ldt.entity.alarm;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @author: lianghanqiang
+ * @description:
+ * @since: 10/28/21 -- 5:04 PM
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class IPAlarm {
+	String ip;
+	Long createTime;
+	Integer times;
+}

ldt-common/src/main/java/com/ldt/entity/alarm/MethodAlarm.java

@@ -0,0 +1,10 @@
+package com.ldt.entity.alarm;
+
+/**
+ * @author: lianghanqiang
+ * @description:
+ * @since: 10/28/21 -- 5:04 PM
+ */
+public class MethodAlarm {
+
+}

ldt-core/pom.xml

@@ -269,6 +269,12 @@
             <artifactId>drools-compiler</artifactId>
             <version>7.10.0.Final</version>
         </dependency>
+        <dependency>
+            <groupId>org.springblade</groupId>
+            <artifactId>ldt-common</artifactId>
+            <version>2.8.2.RELEASE</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

ldt-core/src/main/java/org/springblade/common/config/mq/RabbitConfig.java

@@ -1,8 +1,7 @@
 package org.springblade.common.config.mq;
 
-import org.springframework.amqp.core.Binding;
-import org.springframework.amqp.core.BindingBuilder;
-import org.springframework.amqp.core.DirectExchange;
+
+import com.ldt.consts.QueueConst;
 import org.springframework.amqp.core.Queue;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -15,24 +14,23 @@ import org.springframework.context.annotation.Configuration;
 @Configuration
 public class RabbitConfig {
 
-	@Bean
-	public Queue TestDirectQueue() {
-		return new Queue("LogDirectQueue", true);
-	}
 
+	/**
+	 * 	api log 接口请求日志输出队列
+	 * */
 	@Bean
-	DirectExchange TestDirectExchange() {
-		return new DirectExchange("LogDirectExchange", true, false);
+	public Queue apiLogQueue() {
+		return new Queue(QueueConst.apiLog, true);
 	}
 
-	@Bean
-	Binding bindingDirect() {
-		return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("LogDirectRouting");
-	}
 
+	/**
+	 * 	err log 错误日志输出队列
+	 * */
 	@Bean
-	DirectExchange lonelyDirectExchange() {
-		return new DirectExchange("lonelyDirectExchange");
+	public Queue errLogQueue() {
+		return new Queue(QueueConst.errorLog, true);
 	}
+
 }
 

ldt-core/src/main/java/org/springblade/gateway/shop_gateway/service/impl/AppShopServiceImpl.java

@@ -169,7 +169,6 @@ public class AppShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements
 		mallApp.setAppIdType("MINI_PROGRAM");
 
 		configAppDto.setAppIdList(new ConfigAppDto.AppInfo[]{clientApp, shopApp, mallApp});
-
 		return yeepayCommonService.configApp(configAppDto);
 	}
 

ldt-core/src/main/java/org/springblade/risk/alarm/RiskAlarmer.java

@@ -0,0 +1,19 @@
+package org.springblade.risk.alarm;
+
+import com.ldt.consts.QueueConst;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author: lianghanqiang
+ * @description:
+ * @since: 10/28/21 -- 9:47 AM
+ */
+@Component
+public class RiskAlarmer {
+
+	@RabbitListener(queues = {QueueConst.ipAlarm})
+	public void ipAlertHandle(String data){
+		System.out.println(data);
+	}
+}

ldt-core/src/main/java/org/springblade/risk/log/listener/LdtApiLogListener.java

@@ -0,0 +1,42 @@
+package org.springblade.risk.log.listener;
+
+import cn.hutool.core.bean.BeanUtil;
+import com.alibaba.fastjson.JSON;
+import com.ldt.consts.QueueConst;
+import com.ldt.entity.ApiLogData;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springblade.core.log.event.ApiLogEvent;
+import org.springblade.core.log.model.LogApi;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+
+/**
+ * @author: lianghanqiang
+ * @description: api 日志收集
+ * @since: 10/27/21 -- 2:31 PM
+ */
+
+@Slf4j
+@Component
+@AllArgsConstructor
+public class LdtApiLogListener {
+
+	private RabbitTemplate rabbitTemplate;
+
+	@Async
+	@EventListener({ApiLogEvent.class})
+	public void saveApiLog(ApiLogEvent event) {
+
+		Map<String, Object> source = (Map<String, Object>) event.getSource();
+		LogApi logApi = (LogApi) source.get("log");
+		ApiLogData apiLogData = BeanUtil.toBean(logApi, ApiLogData.class);
+		rabbitTemplate.send(QueueConst.apiLog,new Message(JSON.toJSONString(apiLogData).getBytes()));
+
+	}
+}

ldt-core/src/main/java/org/springblade/risk/log/listener/LdtLogListener.java → ldt-core/src/main/java/org/springblade/risk/log/listener/LdtErrLogListener.java

@@ -3,8 +3,7 @@ package org.springblade.risk.log.listener;
 import com.alibaba.fastjson.JSON;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springblade.core.log.event.UsualLogEvent;
-import org.springframework.amqp.core.Message;
+import org.springblade.core.log.event.ErrorLogEvent;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.context.event.EventListener;
 import org.springframework.scheduling.annotation.Async;
@@ -12,21 +11,21 @@ import org.springframework.stereotype.Component;
 
 /**
  * @author: lianghanqiang
- * @description:  系统自定义日志事件监听---日志收集
- * @since: 10/22/21 -- 3:05 PM
+ * @description: 错误 日志收集
+ * @since: 10/27/21 -- 2:57 PM
  */
+
 @Slf4j
 @Component
 @AllArgsConstructor
-public class LdtLogListener {
+public class LdtErrLogListener {
 
-	private  RabbitTemplate rabbitTemplate;
+	private RabbitTemplate rabbitTemplate;
 
 	@Async
-	@EventListener({UsualLogEvent.class})
-	public void saveApiLog(UsualLogEvent event) {
+	@EventListener({ErrorLogEvent.class})
+	public void saveApiLog(ErrorLogEvent event) {
 		String data = JSON.toJSONString(event.getSource());
-		rabbitTemplate.send("LogDirectExchange","LogDirectRouting",new Message(data.getBytes()));
 		System.out.println(data);
 	}
 }

ldt-core/src/main/java/org/springblade/yeePay/controller/YeepayController.java

@@ -179,6 +179,7 @@ public class YeepayController {
 
 	@PostMapping("/queryOrder")
 	@ApiOperation("查询订单")
+	@ApiLog
 	public JSONObject queryOrder(TradeOrderQueryDto orderQueryDto){
 		return JSON.parseObject(yeepaySaasService.tradeOrderQuery(orderQueryDto).getStringResult());
 	}

ldt-core/src/main/resources/wxConfig.properties

@@ -30,6 +30,8 @@ wx.shop.appName = 联兑通商家
 wx.mall.appId = wxdb5d1dcc471a46d4
 wx.mall.appSecret = aac0f305b9358fc03e19c3b3bb4e24e0
 
+
+
 wx.common.domain = https://ldt.guosen-fumao.cn/api/
 wx.common.businessUrls[0] = https://elecsign.yeepay.com/
 wx.common.businessUrls[1] = http://img.yeepay.com/

ldt-core/src/test/java/org/springblade/test/BladeTest.java

@@ -15,10 +15,10 @@ import org.springblade.core.test.BladeSpringRunner;
 @BladeBootTest(appName = "blade-runner")
 public class BladeTest {
 
-	@Test
-	public void contextLoads() {
-
-
-	}
+//	@Test
+//	public void contextLoads() {
+//
+//
+//	}
 
 }

ldt-risk/pom.xml

@@ -68,6 +68,14 @@ under the License.
             <version>${flink.version}</version>
         </dependency>
 
+        <!--  Flink CEP 事件处理  -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-cep_2.11</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+
+
 
 
         <!-- Add connector dependencies here. They must be in the default scope (compile). -->
@@ -108,6 +116,7 @@ under the License.
             <artifactId>mongo-java-driver</artifactId>
             <version>${mongo.version}</version>
         </dependency>
+
         <dependency>
             <groupId>org.mongodb</groupId>
             <artifactId>bson</artifactId>
@@ -131,6 +140,16 @@ under the License.
             <artifactId>fastjson</artifactId>
             <version>1.2.41</version>
         </dependency>
+        <dependency>
+            <groupId>org.springblade</groupId>
+            <artifactId>blade-starter-log</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springblade</groupId>
+            <artifactId>ldt-common</artifactId>
+            <version>2.8.2.RELEASE</version>
+            <scope>compile</scope>
+        </dependency>
 
     </dependencies>
 

ldt-risk/src/main/java/org/springblade/BatchJob.java

@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springblade;
-
-import org.apache.flink.api.java.ExecutionEnvironment;
-
-/**
- * Skeleton for a Flink Batch Job.
- *
- * <p>For a tutorial how to write a Flink batch application, check the
- * tutorials and examples on the <a href="http://flink.apache.org/docs/stable/">Flink Website</a>.
- *
- * <p>To package your application into a JAR file for execution,
- * change the main class in the POM.xml file to this class (simply search for 'mainClass')
- * and run 'mvn clean package' on the command line.
- */
-public class BatchJob {
-
-	public static void main(String[] args) throws Exception {
-		// set up the batch execution environment
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
-		/*
-		 * Here, you can start creating your execution plan for Flink.
-		 *
-		 * Start with getting some data from the environment, like
-		 * 	env.readTextFile(textPath);
-		 *
-		 * then, transform the resulting DataSet<String> using operations
-		 * like
-		 * 	.filter()
-		 * 	.flatMap()
-		 * 	.join()
-		 * 	.coGroup()
-		 *
-		 * and many more.
-		 * Have a look at the programming guide for the Java API:
-		 *
-		 * http://flink.apache.org/docs/latest/apis/batch/index.html
-		 *
-		 * and the examples
-		 *
-		 * http://flink.apache.org/docs/latest/apis/batch/examples.html
-		 *
-		 */
-
-		// execute program
-		env.execute("Flink Batch Java API Skeleton");
-	}
-}

ldt-risk/src/main/java/org/springblade/StreamingJob.java

@@ -1,21 +1,3 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
 package org.springblade;
 
 import lombok.extern.slf4j.Slf4j;
@@ -23,68 +5,27 @@ import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
-import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
-import org.springblade.sink.MongoSink;
+import org.springblade.sink.mongodb.MongoSink;
+import org.springblade.utils.RabbitUtils;
+
 
-/**
- * Skeleton for a Flink Streaming Job.
- *
- * <p>For a tutorial how to write a Flink streaming application, check the
- * tutorials and examples on the <a href="http://flink.apache.org/docs/stable/">Flink Website</a>.
- *
- * <p>To package your application into a JAR file for execution, run
- * 'mvn clean package' on the command line.
- *
- * <p>If you change the name of the main class (with the public static void main(String[] args))
- * method, change the respective entry in the POM.xml file (simply search for 'mainClass').
- */
 @Slf4j
 public class StreamingJob {
 
 	public static void main(String[] args) throws Exception {
-		// set up the streaming execution environment
-		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
-		/*
-		 * Here, you can start creating your execution plan for Flink.
-		 *
-		 * Start with getting some data from the environment, like
-		 * 	env.readTextFile(textPath);
-		 *
-		 * then, transform the resulting DataStream<String> using operations
-		 * like
-		 * 	.filter()
-		 * 	.flatMap()
-		 * 	.join()
-		 * 	.coGroup()
-		 *
-		 * and many more.
-		 * Have a look at the programming guide for the Java API:
-		 *
-		 * http://flink.apache.org/docs/latest/apis/streaming/index.html
-		 *
-		 */
-
-		// execute program
-		final RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
-			.setHost("192.168.1.218")
-			.setPort(5672)
-			.setUserName("guest")
-			.setPassword("guest")
-			.setVirtualHost("ldt")
-			.build();
+		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
 		final DataStream<String> stream = env
 			.addSource(new RMQSource<String>(
-				connectionConfig,            // config for the RabbitMQ connection
-				"LogDirectQueue",                 // name of the RabbitMQ queue to consume
-				true,                        // use correlation ids; can be false if only at-least-once is required
-				new SimpleStringSchema()))   // deserialization schema to turn messages into Java objects
+				RabbitUtils.getRMQConnectionConfig(),
+				"LogDirectQueue",
+				true,
+				new SimpleStringSchema()))
 			.setParallelism(1);
 
 		stream.addSink(new MongoSink());
 		stream.print("demo");
-		log.error("ceshi===========================================================================");
-		env.execute("Flink Streaming Java API Skeleton");
+		env.execute("job");
 	}
 }

ldt-risk/src/main/java/org/springblade/resource/RabbitSource.java

@@ -1,9 +0,0 @@
-package org.springblade.resource;
-
-/**
- * @author: lianghanqiang
- * @description: rabbitmq connector configuration
- * @since: 10/26/21 -- 3:56 PM
- */
-public class RabbitSource {
-}

ldt-risk/src/main/java/org/springblade/sink/mongodb/ApiLogWriter.java

@@ -0,0 +1,40 @@
+package org.springblade.sink.mongodb;
+
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.db.nosql.mongo.MongoFactory;
+import com.alibaba.fastjson.JSON;
+import com.mongodb.client.MongoDatabase;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.bson.Document;
+
+/**
+ * @author: lianghanqiang
+ * @description: api log 输出 sink
+ * @since: 10/27/21 -- 3:22 PM
+ */
+@Slf4j
+public class ApiLogWriter extends RichSinkFunction<String> {
+
+	private MongoDatabase mongoDatabase;
+
+	@Override
+	public void open(Configuration parameters) throws Exception {
+		this.mongoDatabase = MongoFactory.getDS("master").getDb("ldt");
+		super.open(parameters);
+	}
+
+	@Override
+	public void invoke(String value, Context context) {
+		log.info(value);
+		Document document = new Document();
+		document.append("data","data");
+		this.mongoDatabase.getCollection("log").insertOne(document);
+	}
+
+	@Override
+	public void close() throws Exception {
+		super.close();
+	}
+}

ldt-risk/src/main/java/org/springblade/sink/mongodb/IPAlarmSink.java

@@ -0,0 +1,51 @@
+package org.springblade.sink.mongodb;
+
+import cn.hutool.db.nosql.mongo.MongoFactory;
+import com.ldt.entity.alarm.IPAlarm;
+import com.mongodb.client.MongoDatabase;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.bson.Document;
+import org.springblade.utils.MongoUtils;
+
+
+/**
+ * @author: lianghanqiang
+ * @description:
+ * @since: 10/28/21 -- 6:05 PM
+ */
+public class IPAlarmSink  extends RichSinkFunction<IPAlarm> {
+
+	private MongoDatabase mongoDatabase;
+
+	/**
+	 * 链接mongo
+	 */
+	@Override
+	public void open(Configuration parameters) throws Exception {
+		this.mongoDatabase = MongoFactory.getDS("master").getDb(MongoUtils.getProperties().get("master","database"));
+		super.open(parameters);
+	}
+
+	/**
+	 * 将数据写入
+	 */
+	@Override
+	public void invoke(IPAlarm value, Context context) {
+		System.out.println(value);
+		Document document = new Document();
+		document.append("ip",value.getIp());
+		document.append("eventTime",value.getTimes());
+		document.append("frequency",value.getTimes());
+		this.mongoDatabase.getCollection("ip_alarm").insertOne(document);
+	}
+
+	/**
+	 * 释放资源
+	 */
+	@Override
+	public void close() throws Exception {
+		super.close();
+	}
+
+}

ldt-risk/src/main/java/org/springblade/sink/MongoSink.java → ldt-risk/src/main/java/org/springblade/sink/mongodb/MongoSink.java

@@ -1,11 +1,12 @@
-package org.springblade.sink;
+package org.springblade.sink.mongodb;
 
-import com.mongodb.MongoClient;
+import cn.hutool.db.nosql.mongo.MongoFactory;
+import com.ldt.entity.alarm.IPAlarm;
 import com.mongodb.client.MongoDatabase;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 import org.bson.Document;
-import org.springblade.utils.MongoUtils;
 
 
 /**
@@ -22,7 +23,7 @@ public class MongoSink extends RichSinkFunction<String> {
 	 */
 	@Override
 	public void open(Configuration parameters) throws Exception {
-		this.mongoDatabase = MongoUtils.getMongoDatabase("ldt");
+		this.mongoDatabase = MongoFactory.getDS("master").getDb("ldt");
 		super.open(parameters);
 	}
 
@@ -30,7 +31,7 @@ public class MongoSink extends RichSinkFunction<String> {
 	 * 将数据写入
 	 */
 	@Override
-	public void invoke(String value, Context context) throws Exception {
+	public void invoke(String value, Context context) {
 		System.out.println(value);
 		Document document = new Document();
 		document.append("data","data");

ldt-risk/src/main/java/org/springblade/sink/rabbitmq/ObjectJsonParser.java

@@ -0,0 +1,17 @@
+package org.springblade.sink.rabbitmq;
+
+import com.alibaba.fastjson.JSON;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+
+/**
+ * @author: lianghanqiang
+ * @description: rabbitmq sink json 序列化
+ * @since: 10/28/21 -- 5:12 PM
+ */
+public class ObjectJsonParser<T> implements SerializationSchema<T> {
+
+	@Override
+	public byte[] serialize(T t) {
+		return JSON.toJSONString(t).getBytes();
+	}
+}

ldt-risk/src/main/java/org/springblade/transform/ApiLogTransformer.java

@@ -0,0 +1,104 @@
+package org.springblade.transform;
+
+import com.alibaba.fastjson.JSON;
+import com.ldt.consts.QueueConst;
+import com.ldt.entity.ApiLogData;
+import com.ldt.entity.alarm.IPAlarm;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
+import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
+import org.springblade.sink.mongodb.IPAlarmSink;
+import org.springblade.sink.mongodb.MongoSink;
+import org.springblade.sink.rabbitmq.ObjectJsonParser;
+import org.springblade.utils.RabbitUtils;
+
+import java.time.Duration;
+import java.util.Date;
+import java.util.Optional;
+
+
+/**
+ * @author: lianghanqiang
+ * @description: api 日志处理算子
+ * @since: 10/27/21 -- 3:12 PM
+ */
+@Slf4j
+public class ApiLogTransformer {
+
+	public static void main(String[] args) throws Exception {
+
+
+		//获取执行环境
+		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+		//添加 rabbitmq 数据源，获取队列中的数据
+		final DataStream<String> stream = env.addSource(
+			new RMQSource<>(
+				RabbitUtils.getRMQConnectionConfig(),    //rabbitmq 配置环境
+				QueueConst.apiLog,                    //rabbitmq 监听队列名称
+				true,
+				new SimpleStringSchema()))			//序列化规则
+			.setParallelism(1);		//设置任务的并行数量（线程数）
+
+
+		//日志数据
+		SingleOutputStreamOperator<ApiLogData> apiLogData = stream.map(mapString -> JSON.parseObject(mapString, ApiLogData.class));
+
+		/**	高频IP访问告警	*/
+		SingleOutputStreamOperator<IPAlarm> ipAlarms = apiLogData
+			.map(
+				data -> IPAlarm.builder().ip(data.getRemoteIp())
+					.createTime(Optional.ofNullable(data.getCreateTime()).orElse(new Date()).getTime())
+					.times(1)
+					.build()
+			)
+			.assignTimestampsAndWatermarks(
+				WatermarkStrategy
+					.<IPAlarm>forBoundedOutOfOrderness(Duration.ofSeconds(10))
+					.withTimestampAssigner((data, timestamp) -> data.getCreateTime())
+			)
+			.keyBy(data -> data.getIp())
+			.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
+			.sum("times")
+			.filter(data -> data.getTimes() > 5);
+
+		ipAlarms.addSink(new RMQSink<IPAlarm>(RabbitUtils.getRMQConnectionConfig(),QueueConst.ipAlarm, new ObjectJsonParser()));
+		ipAlarms.addSink(new IPAlarmSink());
+
+		/**	高频接口调用告警	*/
+		SingleOutputStreamOperator<Tuple3<String, Long, Integer>> methodAlarm = apiLogData
+			.map(
+				(MapFunction<ApiLogData, Tuple3<String, Long, Integer>>) apiLogData1 -> new Tuple3(
+					apiLogData1.getRequestUri(),
+					Optional.ofNullable(apiLogData1.getCreateTime()).orElse(new Date()).getTime(),
+					1)
+			)
+			.returns(Types.TUPLE(Types.STRING, Types.LONG, Types.INT))
+//			.assignTimestampsAndWatermarks(
+//				WatermarkStrategy
+//					.<Tuple3<String, Long, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(10))
+//					.withTimestampAssigner((data, timestamp) -> data.f1)
+//			)
+			.keyBy(data -> data.f0)
+			.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
+			.sum(2)
+			.filter(data -> data.f2 > 5);
+/*
+		ipAlarms.addSink(new RMQSink<IPAlarm>(RabbitUtils.getRMQConnectionConfig(),QueueConst.ipAlarm, new ObjectJsonParser()));
+		ipAlarms.addSink(new IPAlarmSink());*/
+
+		//执行job
+		env.execute("ApiLogJob");
+
+	}
+}

ldt-risk/src/main/java/org/springblade/utils/MongoUtils.java

@@ -1,70 +1,15 @@
 package org.springblade.utils;
 
-import cn.hutool.core.io.resource.ClassPathResource;
-import com.alibaba.fastjson.JSON;
-import com.mongodb.MongoClient;
-import com.mongodb.MongoCredential;
-import com.mongodb.ServerAddress;
-import com.mongodb.client.MongoDatabase;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.Properties;
+import cn.hutool.setting.Setting;
 
 /**
  * @author: lianghanqiang
- * @description: mongo 连接工具
- * @since: 10/26/21 -- 3:53 PM
+ * @description: Rabbitmq 配置信息封装工具
+ * @since: 10/28/21 -- 6:09 PM
  */
-@Data
-@Slf4j
 public class MongoUtils {
 
-	private static MongoClient mongoClient;
-	private static MongoDatabase mongoDatabase;
-
-
-	public static MongoDatabase getMongoDatabase(String databaseName){
-		if(Objects.isNull(mongoClient)){
-			ClassPathResource resource = new ClassPathResource("mongo.properties");
-			Properties properties = new Properties();
-			try {
-				properties.load(resource.getStream());
-
-				String host = properties.getProperty("host");
-				int port = Integer.parseInt(properties.getProperty("port","27017"));
-				String userName = properties.getProperty("userName");
-				String password = properties.getProperty("password");
-				String database = properties.getProperty("database");
-
-				//通过连接认证获取MongoDB连接
-				ServerAddress serverAddress = new ServerAddress(host,port);
-				List<ServerAddress> addrs = new ArrayList<ServerAddress>();
-				addrs.add(serverAddress);
-
-				//MongoCredential.createScramSha1Credential()三个参数分别为 用户名 数据库名称 密码
-				MongoCredential credential = MongoCredential.createScramSha1Credential(userName, database, password.toCharArray());
-				List<MongoCredential> credentials = new ArrayList<MongoCredential>();
-				credentials.add(credential);
-
-				//通过连接认证获取MongoDB连接
-				mongoClient = new com.mongodb.MongoClient(addrs,credentials);
-				System.out.println(JSON.toJSONString(properties));
-			} catch (IOException e) {
-				log.error("============================== error start ( Mongo connect error ) =====================");
-				e.printStackTrace();
-				log.error("============================== error end   ( Mongo connect error ) =====================");
-			}
-		}
-		return mongoClient.getDatabase(databaseName);
-	}
-
-
-	public static void main(String[] args){
-		MongoDatabase ldt = MongoUtils.getMongoDatabase("ldt");
-		System.out.println(ldt);
+	public static Setting getProperties(){
+		return new Setting("config/mongo.setting");
 	}
 }

ldt-risk/src/main/java/org/springblade/utils/RabbitUtils.java

@@ -0,0 +1,24 @@
+package org.springblade.utils;
+
+import cn.hutool.setting.Setting;
+import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
+
+/**
+ * @author: lianghanqiang
+ * @description: Rabbitmq 配置信息封装工具
+ * @since: 10/27/21 -- 2:01 PM
+ */
+public class RabbitUtils {
+
+	public static RMQConnectionConfig getRMQConnectionConfig(){
+		Setting setting = new Setting("config/rabbit.setting");
+		return new RMQConnectionConfig.Builder()
+			.setHost(setting.get("host"))
+			.setPort(setting.getInt("port"))
+			.setUserName(setting.get("user"))
+			.setPassword(setting.get("pass"))
+			.setVirtualHost(setting.get("virtualHost"))
+			.build();
+	}
+
+}

ldt-risk/src/main/resources/config/mongo.setting

@@ -0,0 +1,20 @@
+#每个主机答应的连接数（每个主机的连接池大小），当连接池被用光时，会被阻塞住 ，默以为10 --int
+connectionsPerHost=100
+#线程队列数，它以connectionsPerHost值相乘的结果就是线程队列最大值。如果连接线程排满了队列就会抛出“Out of semaphores to get db”错误 --int
+threadsAllowedToBlockForConnectionMultiplier=10
+#被阻塞线程从连接池获取连接的最长等待时间（ms） --int
+maxWaitTime = 120000
+#在建立（打开）套接字连接时的超时时间（ms），默以为0（无穷） --int
+connectTimeout=0
+#套接字超时时间;该值会被传递给Socket.setSoTimeout(int)。默以为0（无穷） --int
+socketTimeout=0
+#是否打开长连接. defaults to false --boolean
+socketKeepAlive=false
+
+#---------------------------------- MongoDB实例连接
+[master]
+host = 192.168.1.218:27017
+user=root
+pass=123456
+database=ldt
+#-----------------------------------------------------

ldt-risk/src/main/resources/config/rabbit.setting

@@ -0,0 +1,5 @@
+host=192.168.1.218
+user=guest
+pass=guest
+virtualHost=ldt
+port=5672

ldt-risk/src/main/resources/mongo.properties

@@ -1,10 +0,0 @@
-#连接端口
-port=27017
-#用户名
-userName=root
-#密码
-password=123456
-#数据库
-database=ldt
-#主机地址
-host=192.168.1.218

pom.xml

@@ -11,6 +11,7 @@
     <modules>
         <module>ldt-core</module>
         <module>ldt-risk</module>
+        <module>ldt-common</module>
     </modules>
 
     <properties>