7

私は、Apache Camel での parallelConsumers とスレッドの概念を理解するのにほぼ 2 日を費やしました。しかし、私はその概念を本当に理解していません。誰でも私が概念を理解するのを手伝ってくれますか? 私はキャメル2.12.0を使用しています。

   from("jms:queue:start?concurrentConsumers=5")
   .threads(3, 3, "replyThread")
   .bean(new SomeBean());

   pom.xml
   ==========================================  
   <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <camel.version>2.12.0</camel.version>
</properties>

<dependencies>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.8.1</version>
        <scope>test</scope>
    </dependency>

    <!-- required by both client and server -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-core</artifactId>
        <version>${camel.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-spring</artifactId>
        <version>${camel.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-jms</artifactId>
        <version>${camel.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-test</artifactId>
        <version>${camel.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-activemq</artifactId>
        <version>1.1.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-test-spring</artifactId>
        <version>2.12.1</version>
    </dependency>
</dependencies>


   //posting to queue
   for (int i = 0; i < 10; i++) {
      System.out.println(Thread.currentThread() + " sending request..." + i);
      template.asyncCallbackRequestBody("jms:queue:start", (body + " " + i), callback);
   } 

   //my call back
   public class MyCallback extends SynchronizationAdapter {

   @Override
   public void onComplete(Exchange exchange) {
      String body = exchange.getOut().getBody(String.class);
      System.out.println(Thread.currentThread() + " Callback Resposne..." +body);
 }
}

  public static class SomeBean {
    public void someMethod(Exchange body) {
        System.out.println(Thread.currentThread() + " Received: " + body.getIn().getBody());
        body.getOut().setBody(body.getIn().getBody());
    }
}

私の理解では(apache camel documentationから)、メッセージを消費するためにjms:startキューに5つの競合する消費者がいるということです。jms:start キューからの非同期応答を処理するための 3 つの「replyThreads」。しかし、実際の出力は異なります。

     Thread[main,5,main] sending request...0
     Thread[main,5,main] sending request...1
     Thread[Camel (camel-1) thread #9 - replyThread,5,main] Received: Hello Camel 0
     Thread[Camel (camel-1) thread #10 - replyThread,5,main] Received: Hello Camel 1
     Thread[Camel (camel-1) thread #5 - ProducerTemplate,5,main] Callback  Resposne...Hello Camel 0
     Thread[Camel (camel-1) thread #6 - ProducerTemplate,5,main] Callback Resposne...Hello Camel 1
4

2 に答える 2

8

JMS コンポーネントには組み込みのスレッド プーリングがあり、キューに入れられたメッセージの数に応じて非常に適切にスケールアップおよびスケールダウンします。

だからそれを使うだけ

 from("jms:queue:start?concurrentConsumers=5")
   .bean(new SomeBean());

最大値も指定できるので、範囲があります

 from("jms:queue:start?concurrentConsumers=5&maxConcurrentConsumers=10")
   .bean(new SomeBean());
于 2013-11-15T10:44:17.987 に答える