java - 如何使用camel "direct-vm"在两个camel上下文之间进行通信？

Question

我正在使用 direct-vm 在同一 JVM 中运行的两个 Camel 上下文之间进行通信。但它不起作用，我的制作人一直在寻找消费者:(我在这里错过了什么吗。感谢任何帮助:)

这是我的生产者，将数据从文件夹中的文件发送到端点

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

public class OrderRouter {

    public static void main(String args[]) throws Exception {
   // create CamelContext
    CamelContext context = new DefaultCamelContext();

// add our route to the CamelContext
    context.addRoutes(new RouteBuilder() {
    @Override
    public void configure() {

        from("file:src/data?noop=true").to("direct-vm:pipeRequestDR91");

        }
    });

    // start the route and let it do its work
    context.start();
    Thread.sleep(10000);

    // stop the CamelContext
    context.stop();
    } 
}

这是我在端点上进行的消费者轮询。

import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor; 
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

public class App {
    public static void main(String[] args) throws Exception {

    CamelContext context = new DefaultCamelContext();

    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() {

            from("direct-vm:pipeRequestDR91").process(new Processor() {
                public void process(Exchange exchange) throws Exception {
                    System.out.println("Reached here");
                    System.out.println("Received XML order: "
                            + exchange.getIn().getHeader("CamelFileName"));
                    String strBody = exchange.getIn().getBody(String.class);
                    System.out.println(strBody);
                }
            });

        }
    });

    context.start();
    Thread.sleep(100000);
    }
}

这是错误:

    Message History

---------------------------------------------------------------------------------------------------------------------------------------
RouteId              ProcessorId          Processor                                                                        Elapsed (ms)
[route1            ] [route1            ] [file://src/data?noop=true                                                     ] [         5]
[route1            ] [to1               ] [direct-vm:pipeRequestDR91                                                     ] [         0]

Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
    Id                  ID-LP0G0085-56374-1502833874430-0-418
    ExchangePattern     InOnly
    Headers             {breadcrumbId=ID-LP0G0085-56374-1502833874430-0-417, CamelFileAbsolute=false, CamelFileAbsolutePath=C:\Prashant\camelinaction-master\chapter2\cbr\src\data\message2.csv, CamelFileContentType=application/vnd.ms-excel, CamelFileLastModified=1436197636000, CamelFileLength=53, CamelFileName=message2.csv, CamelFileNameConsumed=message2.csv, CamelFileNameOnly=message2.csv, CamelFileParent=src\data, CamelFilePath=src\data\message2.csv, CamelFileRelativePath=message2.csv, CamelRedelivered=false, CamelRedeliveryCounter=0}
    BodyType            org.apache.camel.component.file.GenericFile
    Body                [Body is file based: GenericFile[message2.csv]]
]

Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.component.directvm.DirectVmConsumerNotAvailableException: No consumers available on endpoint: Endpoint[direct-vm://pipeRequestDR91]. Exchange[message2.csv]
    at org.apache.camel.component.directvm.DirectVmProducer.process(DirectVmProducer.java:51)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:129)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:435)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:211)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:175)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:174)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.impl.ScheduledPollConsumer.run(ScheduledPollConsumer.java:101)[camel-core-2.15.2.jar:2.15.2]
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)[:1.7.0_71]
    at java.util.concurrent.FutureTask.runAndReset(Unknown Source)[:1.7.0_71]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(Unknown Source)[:1.7.0_71]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)[:1.7.0_71]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)[:1.7.0_71]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)[:1.7.0_71]
    at java.lang.Thread.run(Unknown Source)[:1.7.0_71]
[1) thread #0 - file://src/data] GenericFileOnCompletion        WARN  Rollback file strategy: org.apache.camel.component.file.strategy.GenericFileRenameProcessStrategy@281eff for file: GenericFile[message3.csl]
[1) thread #0 - file://src/data] DefaultErrorHandler

Answer 1

这是一个最小的示例(但是 Thread.Sleep 只是一种快速方法来显示同一 jvm 中的两个上下文意味着什么，而不需要 osgi/spring 等):

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

public class Main {
    public static void main(String[] args) throws Exception {
        CamelContext first = new DefaultCamelContext();
        first.addRoutes(new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("timer:start").to("direct-vm:test");
            }
        });

        CamelContext second = new DefaultCamelContext();
        second.addRoutes(new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct-vm:test").log("Message received");
            }
        });

        first.start();
        second.start();

        Thread.sleep(100000);
    }
}

实际上，据我所知，真正的需求是在不同应用程序中的不同 Camel 上下文之间进行通信，因此 JVM 会有所不同。与“direct”或“direct-vm”类似的最快方法是使用camel-netty4并将选项“transferExchange”设置为“true”( http://camel.apache.org/netty4.html )。下面是一个示例，其中包含两个使用这种方式进行通信的独立应用程序:

App1.java

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

public class App1 {
    public static void main(String[] args) throws Exception {
        CamelContext first = new DefaultCamelContext();
        first.addRoutes(new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("timer:start").to("netty4:tcp://127.0.0.1:9999?transferExchange=true");
            }
        });

        first.start();
        Thread.sleep(100000);
    }
}

App2.java

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

public class App2 {
    public static void main(String[] args) throws Exception {
        CamelContext second = new DefaultCamelContext();
        second.addRoutes(new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("netty4:tcp://127.0.0.1:9999?transferExchange=true").log("Message received");
            }
        });

        second.start();

        Thread.sleep(100000);
    }
}

通过更改 IP 地址，可以在不同的计算机等上运行应用程序。