3

最近开始玩 Spring Cloud Stream 和 RabbitMQ binder。

如果我在两个服务想要传递消息时正确理解所有内容，一个应该配置source来发送消息，另一个应该配置sink来接收消息 - 两者都应该使用相同的channel。

我有一个名为testchannel. 不过，我注意到该源创建了 RabbitMQ 绑定：

  • 交换testchannel，
  • 路由键testchannel，
  • 队列testchannel.default（持久），

而sink创建了 RabbitMQ 绑定：

  • 交换testchannel，
  • 路由键#，
  • 队列testchannel.anonymous.RANDOM_ID（独家）。

为简洁起见，我跳过了前缀。

现在，当我运行这两个应用程序时。第一个将消息发送到testchannel交换，然后将其路由到两个队列（我假设路由键是testchannel）。第二个应用程序使用随机队列中的消息，但从不使用默认队列中的消息。

我的另一个问题是 - 第二个应用程序只使用sink，但它也为输出通道创建绑定，output默认情况下，因为我没有指定任何东西。

我使用相同的 Gradle 脚本构建这两个应用程序：

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

apply plugin: 'java'
apply plugin: 'spring-boot'

repositories {
    mavenCentral()
    maven { url 'https://repo.spring.io/snapshot' }
    maven { url 'https://repo.spring.io/milestone' }
}

dependencies {
    compile(
            'org.springframework.cloud:spring-cloud-starter-stream-rabbit',
    )
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:Brixton.BUILD-SNAPSHOT"
    }
}

第一个应用程序属性：

server.port=8010
spring.cloud.stream.binder.rabbit.default.prefix=z.
spring.cloud.stream.bindings.input=start
spring.cloud.stream.bindings.output=testchannel
spring.rabbitmq.addresses=host1:5672,host2:5672
spring.rabbitmq.username=user
spring.rabbitmq.password=psw

第一个应用程序源代码：

@EnableBinding(Processor.class)
...
@ServiceActivator(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public byte[] handleIncomingMessage(byte[] payload) {}

第二个应用程序属性：

server.port=8011
spring.cloud.stream.binder.rabbit.default.prefix=z.
spring.cloud.stream.bindings.input=testchannel
spring.rabbitmq.addresses=host1:5672,host2:5672
spring.rabbitmq.username=user
spring.rabbitmq.password=psw

第二个应用程序源代码：

@EnableBinding(Sink.class)
...
@ServiceActivator(inputChannel = Sink.INPUT)
public void handleIncomingMessage(byte[] payload) {}

所以我的问题是。

  • 源和接收器不应该使用相同的通道并因此使用相同的代理队列吗？实现这一目标的正确配置是什么？（我的目标是拥有多个接收器服务实例，但只有一个应该使用该消息。）
  • 当我只使用sink时，框架应该创建输出绑定吗？如果是，如何禁用它。
4

1 回答 1

2

默认; 每个消费者都有自己的队列；这是一个发布/订阅方案。

有一个消费者的概念，group因此您可以让多个实例竞争来自同一个队列的消息。

绑定生产者时，会绑定一个默认队列。

如果您想订阅该default群组；你必须设置组：

spring.cloud.stream.bindings.input.group=default

如果您不提供组，您将获得一个独占的自动删除队列。

编辑

由于默认队列是持久的，因此您还应该设置

spring.cloud.stream.bindings.input.durableSubscription=true

避免在消费者绑定时发出警告，并确保在消费者先绑定且队列尚不存在的情况下队列是持久的。

于 2016-02-05T19:44:07.437 回答