Spring Stream尝试在IBM MQueue中创建主题而不是队列

Question

我正在尝试弄清楚如何使用Spring Boot Streams和使用spring-cloud-stream-binder-ibm-mq的MQueue。我可以连接到MQueue，但需要一个Could not provision topic 'queue///EMB_DEV_QUEUE'和MQJE001: Completion Code '2', Reason '2035'。我确实向管理员确认了这是一个队列，而不是一个主题。

我可以使用基于simplest-sample-applications-using-websphere-mq-jms的MQQueueConnectionFactory使用一些示例代码进行连接，因此我知道MQueue正在工作。

这是我的程序。我已经成功地对Kafka使用了相同的模式。

@EnableBinding({Sink.class, Source.class})
@SpringBootApplication
public class MQueueStreamApplication {

    private final static AtomicInteger counter = new AtomicInteger();
    private final        Logger        logger  = LoggerFactory.getLogger(getClass());

    public static void main(String[] args) {
        SpringApplication.run(MQueueStreamApplication.class, args);
    }

    @Bean
    @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedRate = "2000"))
    public MessageSource<String> timeSource() {
        return () -> {
            String message = String.format("Timed Message %d", counter.incrementAndGet());

            logger.info("Producing Message: {}", message);

            return MessageBuilder.withPayload(message).setHeader("Message-Type", "mqueue-stream").build();
        };
    }

    @ServiceActivator(inputChannel = Sink.INPUT)
    public void serviceSink(Message<String> message) {
        String payload = String.valueOf(message.getPayload());

        logger.info("Received Message: {} [{}]", payload, message.getHeaders());
    }

}

这是我的application.yml。我尝试过使用和不使用queue:///前缀。示例程序使用前缀。

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: queue:///EMB_DEV_QUEUE
          group: mqueue-stream
#          binder: ibmmq
        output:
          destination: queue:///EMB_DEV_QUEUE

ibmmq:
  host: vm-dev-q01.corp.int
  port: 1414
  channel: EMB_DEV_CHANNEL
  queueManager: EMB_DEV_QMGR

这是我的Gradle版本。

buildscript {
    ext {
        springBootVersion = '1.5.3.RELEASE'
    }
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
    }
}

apply plugin: 'java'
apply plugin: 'eclipse'
apply plugin: 'idea'
apply plugin: 'org.springframework.boot'

version = '0.0.1-SNAPSHOT'

sourceCompatibility = 1.8
targetCompatibility = 1.8

repositories {
    mavenLocal()
    mavenCentral()
}


dependencies {
    compile('org.springframework.boot:spring-boot-starter-actuator')

    compile('org.springframework.cloud:spring-cloud-stream')
    compile('org.springframework.cloud:spring-cloud-stream-binder-jms-ibm-mq:1.0.0.BUILD-SNAPSHOT')

    testCompile('org.springframework.boot:spring-boot-starter-test')
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:Dalston.RELEASE"
    }
}

我按照说明构建了spring-cloud-stream-binder-ibm-mq。我从MQueue安装程序中获得了所需的两个jars。清单上写着9.0.0.0版本，所以我在pom.xml中使用了9

我是MQueue的新手，对Streams的使用经验有限。我已经成功连接到Kafka了。如果有任何帮助，我将不胜感激。

韦斯。

Answer 1

Spring Cloud Stream使用比常规JMS/IBM-MQ应用程序更自以为是的基础设施，以便能够实现诸如消费者组和分区之类的功能-在本例中，目的地是一个主题-有关详细信息，请参阅https://github.com/spring-cloud/spring-cloud-stream-binder-ibm-mq#how-it-works。