业务功能：根据告警规则，从告警流中将主告警和次告警进行关联。

pom maven配置：

<drools.version>6.5.0.Final</drools.version>
<flink.version>1.10.0</flink.version><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version><!--            <scope>provided</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_${scala.binary.version}</artifactId><version>${flink.version}</version><!--            <scope>provided</scope>--></dependency><dependency><groupId>org.drools</groupId><artifactId>drools-core</artifactId><version>${drools.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.drools/drools-compiler --><dependency><groupId>org.drools</groupId><artifactId>drools-compiler</artifactId><version>${drools.version}</version></dependency><dependency><groupId>org.drools</groupId><artifactId>drools-decisiontables</artifactId><version>${drools.version}</version></dependency><dependency><groupId>org.drools</groupId><artifactId>drools-templates</artifactId><version>${drools.version}</version></dependency><dependency><groupId>org.kie</groupId><artifactId>kie-api</artifactId><version>${drools.version}</version></dependency>

1、普通的Java 的POJO(对应Drools中的fact 对象)

package www.lxk.com;import java.io.Serializable;
import java.util.Date;public class Alarm implements Serializable{private String alarmId;private Date eventTime;private String relatedType;public String getAlarmId() {return alarmId;}public void setAlarmId(String alarmId) {this.alarmId = alarmId;}public Date getEventTime() {return eventTime;}public void setEventTime(Date eventTime) {this.eventTime = eventTime;}public String getRelatedType() {return relatedType;}public void setRelatedType(String relatedType) {this.relatedType = relatedType;}@Overridepublic String toString() {return "Alarm{" +"alarmId='" + alarmId + '\'' +", eventTime=" + eventTime +", relatedType='" + relatedType + '\'' +'}';}
}

2、规则文件rules2.drl

package rulesimport com.asiainfo.aisware.oss.sink.Alarm
import java.util.ArrayListglobal java.util.List listdeclare Alarm@role(event)        //声明为事件@timestamp(eventTime)  //事件时间戳@expire(5S)      //事件过期时间10分钟，过期后不再匹配任何规则endrule "rule2"no-loop trueduration 3000when$pAlarm : Alarm( $qq: alarmId memberOf ["111"] ) from entry-point "Demo02"$sList:ArrayList(size >=10 ) from collect (Alarm(alarmId memberOf ["222","333"] )from entry-point "Demo02")then$pAlarm.setRelatedType("主告警");System.out.println("匹配到主告警" + $pAlarm.toString());System.out.println($qq);//drools.insert($pAlarm);$sList.forEach(x -> {((Alarm)x).setRelatedType("子告警");System.out.println("匹配到子告警"  + x.toString());});
end

3、处理类--规则的编译、收集和执行

package www.lxk.com;import com.asiainfo.aisware.oss.sink.Alarm;
import com.asiainfo.aisware.oss.sink.Rule;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;public class Demo02 {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 非常关键，一定要设置启动检查点！！env.enableCheckpointing(5000);env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); //精确一次env.setParallelism(1);//告警流DataStream<Alarm> dataStream = env.addSource(new MyAlarmSource());//规则流DataStream<Rule> ruleStream = env.addSource(new MyRuleSource());//dataStream.print();//ruleStream.print();dataStream.connect(ruleStream).flatMap(new RuleAlarmFunction()).print();env.execute("Drools Demo01");}//自定义类，继承SourceFunction，每条数据是一个规则对象（根据业务自定义）public static class MyAlarmSource implements SourceFunction<Alarm>{// 定义一个标识，表示数据源是否继续运行private Boolean running = true;private static AtomicInteger counter = new AtomicInteger(0);@Overridepublic void run(SourceContext<Alarm> sourceContext) throws Exception {String[] ids = {"111","222","333"};while (running){Alarm alarm = new Alarm();alarm.setAlarmId(ids[counter.getAndIncrement()%3]);alarm.setEventTime(new Date());Thread.sleep(500);sourceContext.collect(alarm);}}@Overridepublic void cancel() {running = false;}}//自定义类，继承SourceFunction，每条数据是一个规则对象（根据业务自定义）public static class MyRuleSource implements SourceFunction<Rule>{// 定义一个标识，表示数据源是否继续运行private boolean running = true;@Overridepublic void run(SourceContext<Rule> sourceContext) throws Exception {while (running){//counter = 1;Thread.sleep(500);Rule rule = new Rule();File file = new File("F:\\04_drools\\fm-rule\\src\\main\\resources\\rules\\rules2.drl");FileReader reader = new FileReader(file);BufferedReader br = new BufferedReader(reader);// 建立一个对象，它把文件内容转成计算机能读懂的语言String line;StringBuffer sb = new StringBuffer();//网友推荐更加简洁的写法while ((line = br.readLine()) != null) {// 一次读入一行数据sb.append(line).append("\n");}rule.setDrlStr(sb.toString());rule.setOperate(1);rule.setName("rule2");sourceContext.collect(rule);}}@Overridepublic void cancel() {this.running = false;}}}

package www.lxk.com;import com.asiainfo.aisware.oss.sink.Alarm;
import com.asiainfo.aisware.oss.sink.Rule;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;
import org.drools.core.impl.KnowledgeBaseImpl;
import org.kie.api.KieBaseConfiguration;
import org.kie.api.KieServices;
import org.kie.api.conf.EventProcessingOption;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.EntryPoint;
import org.kie.internal.builder.KnowledgeBuilder;
import org.kie.internal.builder.KnowledgeBuilderFactory;
import org.kie.internal.io.ResourceFactory;
import org.kie.internal.utils.KieHelper;import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;public class RuleAlarmFunction extends RichCoFlatMapFunction<Alarm, Rule, Alarm> {private  KieHelper kieHelper;private  KnowledgeBaseImpl kieBase;private  KieSession kieSession;private  EntryPoint entryPoint;private Lock lock = new ReentrantLock();@Overridepublic void open(Configuration parameters) throws Exception {if(kieSession == null ){kieHelper = new KieHelper();//kieHelper.addContent(drlStr, ResourceType.DRL);KieBaseConfiguration config = KieServices.Factory.get().newKieBaseConfiguration();config.setOption( EventProcessingOption.STREAM );try{kieBase =(KnowledgeBaseImpl) kieHelper.build();kieSession = kieBase.newStatefulSession();}catch (Exception e){e.printStackTrace();}}}@Overridepublic void close() throws Exception {Thread.sleep(1000);kieSession.destroy();}@Overridepublic void flatMap1(Alarm alarm, Collector collector) throws Exception {lock.lock();try{entryPoint = kieSession.getEntryPoint("Demo02");if(entryPoint != null){entryPoint.insert(alarm);int fireNum = kieSession.fireAllRules();}}catch (Exception e){e.printStackTrace();}finally {lock.unlock();}collector.collect(alarm);}@Overridepublic void flatMap2(Rule rule, Collector collector) throws Exception {if(rule.getOperate() == 0){lock.lock();try{kieBase.removeRule("rules", rule.getName());kieSession = kieBase.newKieSession();}finally {lock.unlock();}}else{// 规则名是ruleIdorg.kie.api.definition.rule.Rule ruleCache = kieBase.getRule("rules", rule.getName());// 规则如果已存在，不需要再次新增if(ruleCache !=null && rule.getName().equals(ruleCache.getName())){System.out.println("规则已经存在，ruleName="+ ruleCache.getName());}else {//重新添加规则KnowledgeBuilder kb = KnowledgeBuilderFactory.newKnowledgeBuilder();//装入规则，可以装入多个kb.add(ResourceFactory.newByteArrayResource(rule.getDrlStr().getBytes("utf-8")), ResourceType.DRL);kieBase.addKnowledgePackages(kb.getKnowledgePackages());kieSession = kieBase.newKieSession();}}}
}

 4、运行测试

"C:\Program Files\Java\jdk1.8.0_201\bin\java" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA...Alarm{alarmId='111', eventTime=Sun Jun 07 17:01:17 CST 2020, relatedType='null'}
规则已经存在，ruleName=rule2
Alarm{alarmId='222', eventTime=Sun Jun 07 17:01:18 CST 2020, relatedType='null'}
规则已经存在，ruleName=rule2
匹配到主告警Alarm{alarmId='111', eventTime=Sun Jun 07 17:01:14 CST 2020, relatedType='主告警'}
111
匹配到子告警Alarm{alarmId='333', eventTime=Sun Jun 07 17:01:09 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='222', eventTime=Sun Jun 07 17:01:09 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='222', eventTime=Sun Jun 07 17:01:10 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='333', eventTime=Sun Jun 07 17:01:11 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='222', eventTime=Sun Jun 07 17:01:12 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='333', eventTime=Sun Jun 07 17:01:12 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='222', eventTime=Sun Jun 07 17:01:13 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='333', eventTime=Sun Jun 07 17:01:14 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='222', eventTime=Sun Jun 07 17:01:15 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='333', eventTime=Sun Jun 07 17:01:15 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='222', eventTime=Sun Jun 07 17:01:16 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='333', eventTime=Sun Jun 07 17:01:17 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='222', eventTime=Sun Jun 07 17:01:18 CST 2020, relatedType='子告警'}
匹配到子告警Alarm{alarmId='333', eventTime=Sun Jun 07 17:01:18 CST 2020, relatedType='子告警'}
匹配到主告警Alarm{alarmId='111', eventTime=Sun Jun 07 17:01:13 CST 2020, relatedType='主告警'}

Process finished with exit code 0