drools - DROOLS 中的单线程执行？

Question

我的应用程序处理多个请求，但我的知识 session 中的规则仅由单线程执行。 例如: 线程1和线程2进入知识 session 的时间间隔为2毫秒 但是线程 1 执行自己的规则，甚至线程 2 的规则也是由线程 1 执行的。 想象一下，如果有 1000 个请求，这意味着每个请求的规则将仅由 1 个线程执行？

在 DROOLS 中我们有什么方法可以防止这种情况并确保规则由多个线程执行吗？

下面是我尝试过的一个小样本测试:

Java 类:

        import java.math.BigDecimal;

        import org.drools.KnowledgeBase;
        import org.drools.KnowledgeBaseFactory;
        import org.drools.builder.KnowledgeBuilder;
        import org.drools.builder.KnowledgeBuilderError;
        import org.drools.builder.KnowledgeBuilderErrors;
        import org.drools.builder.KnowledgeBuilderFactory;
        import org.drools.builder.ResourceType;
        import org.drools.io.ResourceFactory;
        import org.drools.runtime.StatefulKnowledgeSession;
import org.drools.runtime.rule.WorkingMemoryEntryPoint;
import org.springframework.context.support.ClassPathXmlApplicationContext;



        public class DJ_Test {

            public static void main(String[] args) {
                try {
                    System.out.println("In main");
                    // load up the knowledge base
                    KnowledgeBase kbase = readKnowledgeBase();
                    final StatefulKnowledgeSession ksession = kbase.newStatefulKnowledgeSession();
                    final WorkingMemoryEntryPoint entry =ksession.getWorkingMemoryEntryPoint("RequestStream");

                    final Object  obj_1= new Object();
                    Thread t1 = new Thread(){
                        public void run(){System.out.println(Thread.currentThread().getName() + " is running");
                        entry.insert(obj_1);
                        ksession.fireAllRules();
                        System.out.println(Thread.currentThread().getName() + " is terminated");
                        }
                        };

                   final Object  obj_2= new Object();
                   Thread t2 = new Thread(){
                    public void run(){
                    try{
                        Thread.sleep(8000); 
                    }catch(Exception e){

                    }
                    System.out.println(Thread.currentThread().getName() + " is running");
                    entry.insert(obj_2);
                    ksession.fireAllRules();
                    System.out.println(Thread.currentThread().getName() + " is terminated");
                    }
                    };
                  t1.start();
                  t2.start();





                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }

            private static KnowledgeBase readKnowledgeBase() throws Exception {
               /* KnowledgeBuilder kbuilder = KnowledgeBuilderFactory.newKnowledgeBuilder();
                kbuilder.add(ResourceFactory.newClassPathResource("rulesFlow.bpmn"), ResourceType.BPMN2);
                kbuilder.add(ResourceFactory.newClassPathResource("KansasSalesTax.drl"), ResourceType.DRL);
                kbuilder.add(ResourceFactory.newClassPathResource("MissouriSalesTax.drl"), ResourceType.DRL);
                kbuilder.add(ResourceFactory.newClassPathResource("SalesTax.drl"), ResourceType.DRL);
                KnowledgeBuilderErrors errors = kbuilder.getErrors();
                if (errors.size() > 0) {
                    for (KnowledgeBuilderError error: errors) {
                        System.err.println(error);
                    }
                    throw new IllegalArgumentException("Could not parse knowledge.");
                }
                KnowledgeBase kbase = KnowledgeBaseFactory.newKnowledgeBase();
                kbase.addKnowledgePackages(kbuilder.getKnowledgePackages());
                return kbase;*/
                ClassPathXmlApplicationContext serviceContext = new ClassPathXmlApplicationContext( "droolsContext.xml" );
                return (KnowledgeBase) serviceContext.getBean("kbase1");
            }

            public static class DJ_Message {

                public static final int thread_1 = 1;
                public static final int thread_2 = 2;

                private String message;

                private int status;

                public String getMessage() {
                    return this.message;
                }

                public void setMessage(String message) {
                    this.message = message;
                }

                public int getStatus() {
                    return this.status;
                }

                public void setStatus(int status) {
                    this.status = status;
                }

            }

        }

DRL 文件

package com.sample

import com.sample.DroolsTest.Message;

//global CepService cepService;
declare Object
    @role( event )
end


rule "rule_1"
salience 100
    when
        $o : Object() from entry-point RequestStream
    then
        System.out.println( "Rule 1 fired by " + Thread.currentThread().getName() );
        Thread.sleep(5000);

end


rule "rule_2"
salience 80
    when
        $o : Object() from entry-point RequestStream
    then
        System.out.println( "Rule 2 fired by " + Thread.currentThread().getName() );
        Thread.sleep(5000);

end

rule "rule_3"
salience 60
    when
        $o : Object() from entry-point RequestStream
    then
        System.out.println( "Rule 3 fired by " + Thread.currentThread().getName() );
        //cepService.executingThread1();
end

rule "4"
    when
        Message( status == Message.GOODBYE, myMessage : message )
    then
        System.out.println( myMessage );
        //cepService.executingThread2();
end

Answer 1

您可以在多线程环境中使用有状态知识 session 。在您的应用程序启动之前，您必须将“KnowledgeBase”序列化为“file/db”。稍后，每个线程将不会创建自己的“KnowledgeBase”副本，但将从“file/db”中反序列化“KnowledgeBase”。

如果我们不序列化/反序列化“KnowledgeBase”，每个线程在需要时都会尝试加载规则并创建自己的“KnowledgeBase”，最后在某一时刻，如果线程增加，您的应用程序可能最终抛出“java.lan. OutOfMemory permgen”空间错误。因为每个线程都会尝试通过一次又一次地将类加载到内存中来创建自己的知识库副本。