Spring XD無法找到RabbitMQ ConnectionFactory的類定義

Question

我正在從xd-singlenode模式切換到分佈式，使用版本1.3.1-RELEASE。

我在Docker容器中運行Spring XD，其中有Dockerfile similar to the one here。根據XD guide，如果我想使用RabbitMQ作爲我的數據傳輸，那麼我需要配置一個servers.yml文件。我有RabbitMQ在一個單獨的碼頭集裝箱中運行，並將這兩個集裝箱作爲碼頭組成部分鏈接在一起。總之，我配置了一切（我相信是正確的），但是當我在Spring XD中運行多個流時（這是RabbitMQ將起作用，因爲它將充當容器之間的消息傳輸）Spring XD拋出以下異常：

java.lang.NoClassDefFoundError: org/springframework/amqp/rabbit/connection/ConnectionFactory 
    at java.lang.Class.getDeclaredConstructors0(Native Method) ~[na:1.8.0_131] 
    at java.lang.Class.privateGetDeclaredConstructors(Class.java:2671) ~[na:1.8.0_131] 
    at java.lang.Class.getConstructors(Class.java:1651) ~[na:1.8.0_131] 
    at org.springframework.boot.BeanDefinitionLoader.isComponent(BeanDefinitionLoader.java:276) ~[spring-boot-1.2.3.RELEASE.jar:1.2.3.RELEASE] 
    at org.springframework.boot.BeanDefinitionLoader.load(BeanDefinitionLoader.java:158) ~[spring-boot-1.2.3.RELEASE.jar:1.2.3.RELEASE] 
    at org.springframework.boot.BeanDefinitionLoader.load(BeanDefinitionLoader.java:135) ~[spring-boot-1.2.3.RELEASE.jar:1.2.3.RELEASE] 
    at org.springframework.boot.BeanDefinitionLoader.load(BeanDefinitionLoader.java:127) ~[spring-boot-1.2.3.RELEASE.jar:1.2.3.RELEASE] 
    at org.springframework.boot.SpringApplication.load(SpringApplication.java:615) ~[spring-boot-1.2.3.RELEASE.jar:1.2.3.RELEASE] 
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:314) ~[spring-boot-1.2.3.RELEASE.jar:1.2.3.RELEASE] 
    at org.springframework.boot.builder.SpringApplicationBuilder.run(SpringApplicationBuilder.java:139) [spring-boot-1.2.3.RELEASE.jar:1.2.3.RELEASE] 
    at org.springframework.xd.dirt.plugins.spark.streaming.MessageBusConfiguration.createApplicationContext(MessageBusConfiguration.java:86) [spring-xd-dirt-1.3.1.RELEASE.jar:1.3.1.RELEASE] 
    at org.springframework.xd.dirt.plugins.spark.streaming.MessageBusSender.start(MessageBusSender.java:105) [spring-xd-dirt-1.3.1.RELEASE.jar:1.3.1.RELEASE] 
    at org.springframework.xd.spark.streaming.java.ModuleExecutor$1$1.call(ModuleExecutor.java:58) [spring-xd-spark-streaming-1.3.1.RELEASE.jar:1.3.1.RELEASE] 
    at org.springframework.xd.spark.streaming.java.ModuleExecutor$1$1.call(ModuleExecutor.java:53) [spring-xd-spark-streaming-1.3.1.RELEASE.jar:1.3.1.RELEASE] 
    at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreachPartition$1.apply(JavaRDDLike.scala:206) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreachPartition$1.apply(JavaRDDLike.scala:206) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:806) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:806) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1498) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1498) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at org.apache.spark.scheduler.Task.run(Task.scala:64) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203) [spark-core_2.10-1.3.1.jar:1.3.1] 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_131] 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_131] 
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_131] 
Caused by: java.lang.ClassNotFoundException: org.springframework.amqp.rabbit.connection.ConnectionFactory 
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[na:1.8.0_131] 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[na:1.8.0_131] 
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335) ~[na:1.8.0_131] 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[na:1.8.0_131] 
    ... 26 common frames omitted 

那麼，什麼給？ Spring XD是否不具備必要的RabbitMQ依賴關係？

我甚至嘗試手動將spring-messaging jar添加到/opt/spring-xd/xd/lib，其中包含依賴項，但沒有成功。這裏發生了什麼？

編輯：我會拋出我的servers.yml文件額外的上下文。也許有人可以看到我可能做錯：

spring: 
    profiles: container 
xd: 
    transport: rabbit 
    messagebus: 
    local: 
     queueSize:     2147483647 
     polling:      1000 
     executor: 
     corePoolSize:    0 
     maxPoolSize:    200 
     queueSize:     2147483647 
     keepAliveSeconds:   60 
    rabbit: 
     compressionLevel:   1 
      # bus-level property, applies only when 'compress=true' for a stream module 
      # See java.util.zip.Deflater; 1=BEST_SPEED, 9=BEST_COMPRESSION, ... 
     longStringLimit:    8192 
      # Headers longer than this will not be converted to String and will be a 
      # DataInputStream - such headers will NOT be properly converted back on output. 
     default: 
     ackMode:     AUTO 
      # Valid: AUTO (container acks), NONE (broker acks), MANUAL (consumer acks). 
      # Upper case only. 
      # Note: MANUAL requires specialized code in the consuming module and is unlikely to be 
      # used in an XD application. For more information, see 
      # http://docs.spring.io/spring-integration/reference/html/amqp.html#amqp-inbound-ack 
     autoBindDLQ:    false 
     backOffInitialInterval: 1000 
     backOffMaxInterval:  10000 
     backOffMultiplier:   2.0 
     batchBufferLimit:   10000 
     batchingEnabled:   false 
     batchSize:     100 
     batchTimeout:    5000 
     compress:     false 
     concurrency:    1 
     deliveryMode:    PERSISTENT 
     durableSubscription:  false 
     maxAttempts:    3 
     maxConcurrency:   1 
     prefix:     xdbus. 
      # prefix for queue/exchange names so policies (ha, dle etc.) can be applied 
     prefetch:     1 
     replyHeaderPatterns:  STANDARD_REPLY_HEADERS,* 
     republishToDLQ:   false 
      # When false, normal rabbitmq dlq processing; when true, republish to the DLQ with stack trace 
     requestHeaderPatterns:  STANDARD_REQUEST_HEADERS,* 
     requeue:     true 
     transacted:    false 
     txSize:     1 

# redis: 
#  headers: 
      # comma-delimited list of additional header names to transport 
#  default: 
      # default bus properties, if not specified at the module level 
#  backOffInitialInterval: 1000 
#  backOffMaxInterval:  10000 
#  backOffMultiplier:   2.0 
#  concurrency:    1 
#  maxAttempts:    3 
    kafka: 
     brokers:         kafka:9092 
     zkAddress:        zookeeper:2181 
     mode:         embeddedHeaders 
     offsetManagement:      kafkaTopic 
     headers: 
      # comma-delimited list of additional header names to transport 
     socketBufferSize:      2097152 
     offsetStoreTopic:      SpringXdOffsets 
     offsetStoreSegmentSize:     25000000 
     offsetStoreRetentionTime:    60000 
     offsetStoreRequiredAcks:     1 
     offsetStoreMaxFetchSize:     1048576 
     offsetStoreBatchBytes:     16384 
     offsetStoreBatchTime:     1000 
     offsetUpdateTimeWindow:     10000 
     offsetUpdateCount:      0 
     offsetUpdateShutdownTimeout:    2000 
     default: 
     batchSize:     16384 
     batchTimeout:    0 
     replicationFactor:   1 
     concurrency:    1 
     requiredAcks:    1 
     compressionCodec:   none 
     queueSize:     8192 # must be a power of 2 
     maxWait:     100 
     fetchSize:     1048576 
     minPartitionCount:   1 
     durableSubscription:  false 
     syncProducer:    false 
     syncProducerTimeout:  5000 
--- 
#Config for admin 
spring: 
    profiles: admin 
xd: 
    transport: rabbit 
    messagebus: 
    local: 
     queueSize:     2147483647 
     polling:      1000 
     executor: 
     corePoolSize:    0 
     maxPoolSize:    200 
     queueSize:     2147483647 
     keepAliveSeconds:   60 
    rabbit: 
     compressionLevel:   1 
      # bus-level property, applies only when 'compress=true' for a stream module 
      # See java.util.zip.Deflater; 1=BEST_SPEED, 9=BEST_COMPRESSION, ... 
     longStringLimit:    8192 
      # Headers longer than this will not be converted to String and will be a 
      # DataInputStream - such headers will NOT be properly converted back on output. 
     default: 
     ackMode:     AUTO 
      # Valid: AUTO (container acks), NONE (broker acks), MANUAL (consumer acks). 
      # Upper case only. 
      # Note: MANUAL requires specialized code in the consuming module and is unlikely to be 
      # used in an XD application. For more information, see 
      # http://docs.spring.io/spring-integration/reference/html/amqp.html#amqp-inbound-ack 
     autoBindDLQ:    false 
     backOffInitialInterval: 1000 
     backOffMaxInterval:  10000 
     backOffMultiplier:   2.0 
     batchBufferLimit:   10000 
     batchingEnabled:   false 
     batchSize:     100 
     batchTimeout:    5000 
     compress:     false 
     concurrency:    1 
     deliveryMode:    PERSISTENT 
     durableSubscription:  false 
     maxAttempts:    3 
     maxConcurrency:   1 
     prefix:     xdbus. 
      # prefix for queue/exchange names so policies (ha, dle etc.) can be applied 
     prefetch:     1 
     replyHeaderPatterns:  STANDARD_REPLY_HEADERS,* 
     republishToDLQ:   false 
      # When false, normal rabbitmq dlq processing; when true, republish to the DLQ with stack trace 
     requestHeaderPatterns:  STANDARD_REQUEST_HEADERS,* 
     requeue:     true 
     transacted:    false 
     txSize:     1 

# redis: 
#  headers: 
      # comma-delimited list of additional header names to transport 
#  default: 
      # default bus properties, if not specified at the module level 
#  backOffInitialInterval: 1000 
#  backOffMaxInterval:  10000 
#  backOffMultiplier:   2.0 
#  concurrency:    1 
#  maxAttempts:    3 
    kafka: 
     brokers:         kafka:9092 
     zkAddress:        zookeeper:2181 
     mode:         embeddedHeaders 
     offsetManagement:      kafkaTopic 
     headers: 
      # comma-delimited list of additional header names to transport 
     socketBufferSize:      2097152 
     offsetStoreTopic:      SpringXdOffsets 
     offsetStoreSegmentSize:     25000000 
     offsetStoreRetentionTime:    60000 
     offsetStoreRequiredAcks:     1 
     offsetStoreMaxFetchSize:     1048576 
     offsetStoreBatchBytes:     16384 
     offsetStoreBatchTime:     1000 
     offsetUpdateTimeWindow:     10000 
     offsetUpdateCount:      0 
     offsetUpdateShutdownTimeout:    2000 
     default: 
     batchSize:     16384 
     batchTimeout:    0 
     replicationFactor:   1 
     concurrency:    1 
     requiredAcks:    1 
     compressionCodec:   none 
     queueSize:     8192 # must be a power of 2 
     maxWait:     100 
     fetchSize:     1048576 
     minPartitionCount:   1 
     durableSubscription:  false 
     syncProducer:    false 
     syncProducerTimeout:  5000 

--- 
# Rabbit MQ Properties 
# 
# NOTE: sslProperties is mutually exclusive with keyStore, keyStorePassphrase, trustStore, trustStorePassphrase. 
# if you set inline properties, values in the properties file location given by 'sslProperties' will be ignored. 
# 
spring: 
    rabbitmq: 
    addresses: my-domain:5672 
    adminAddresses: http://my-domain:15672 
    nodes: rabbit@my-domain 
    username: myusername 
    password: mypassword 
    virtual_host:/
    useSSL: false 
    sslProperties: 
    ssl: 
    keyStore: 
    keyStorePassphrase: 
    trustStore: 
    trustStorePassphrase: 

編輯2：它看起來像xd-admin和xd-container腳本do not add the messagebus jars到$CLASSPATH後的AdminServerApplication或ContainerServerApplication開始。我不確定這是不是罪魁禍首。

Answer 1

在開始各自應用程序之前，Spring XD啓動腳本爲xd-admin和xd-containerdo not add any of the messagebus jars to the classpath。他們只添加罐子坐在/lib和哈達罐子lib/hadoop27/。

我能夠通過修改腳本以包含在類路徑上必要的罐子解決此問題：

RABBIT_LIB=$APP_HOME/lib/messagebus/rabbit 
if [ -d "$RABBIT_LIB" ]; then 
    for k in "$RABBIT_LIB"/*.jar; do 
    CLASSPATH="$CLASSPATH":"$k" 
    done 
fi 

Answer 2

需要的jar下XD/lib下的子文件夾，根據您選擇的消息總線上...

../../Downloads/spring-xd-1.3.1.RELEASE/xd/lib/messagebus/rabbit 
../../Downloads/spring-xd-1.3.1.RELEASE/xd/lib/messagebus/rabbit/amqp-client-3.6.0.jar 
../../Downloads/spring-xd-1.3.1.RELEASE/xd/lib/messagebus/rabbit/http-client-1.0.0.RELEASE.jar 
../../Downloads/spring-xd-1.3.1.RELEASE/xd/lib/messagebus/rabbit/spring-amqp-1.5.4.RELEASE.jar 
../../Downloads/spring-xd-1.3.1.RELEASE/xd/lib/messagebus/rabbit/spring-integration-amqp-4.2.5.RELEASE.jar 
../../Downloads/spring-xd-1.3.1.RELEASE/xd/lib/messagebus/rabbit/spring-rabbit-1.5.4.RELEASE.jar 
../../Downloads/spring-xd-1.3.1.RELEASE/xd/lib/messagebus/rabbit/spring-xd-messagebus-rabbit-1.3.1.RELEASE.jar 

運行時補充說，目錄到classpath在解析配置。