如何使用spring云流侦听两个RabbitMQ队列

Question

我有一个可以侦听单个RabbitMQ队列的工作应用程序。

但是，当我添加另一个消耗消息的bean并试图将其绑定到另一个队列时，这两个队列都不是在RabbitMQ中创建的，手动创建时也不会从这些队列中使用任何消息。

我创建了一个小kotlin项目来演示这个问题：

@SpringBootApplication
class SpringCloudStreamTwoRabbitConsumersApplication

fun main(args: Array<String>) {
    runApplication<SpringCloudStreamTwoRabbitConsumersApplication>(*args)
}

package com.example.springcloudstreamtworabbitconsumers

import org.springframework.context.annotation.Bean
import org.springframework.messaging.Message
import org.springframework.stereotype.Component
import java.util.function.Consumer

@Component
class Listener1Config {

    @Bean
    fun listener1(): Consumer<Message<String>> {
        return Consumer { input -> println(input) }
    }

}

package com.example.springcloudstreamtworabbitconsumers

import org.springframework.context.annotation.Bean
import org.springframework.messaging.Message
import org.springframework.stereotype.Component
import java.util.function.Consumer

@Component
class Listener2Config {

    @Bean
    fun listener2(): Consumer<Message<String>> {
        return Consumer { input -> println(input) }
    }

}

application.properties：

# Rabbit properties
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

# Listener 1
spring.cloud.stream.bindings.listener1-in-0.destination=exchange1
spring.cloud.stream.bindings.listener1-in-0.group=exchange1-queue
spring.cloud.stream.rabbit.bindings.listener1-in-0.consumer.queueNameGroupOnly=true
spring.cloud.stream.rabbit.bindings.listener1-in-0.consumer.binding-routing-key-delimiter=,
spring.cloud.stream.rabbit.bindings.listener1-in-0.consumer.bindingRoutingKey=binding.key.1,binding.key.1.1
spring.cloud.stream.rabbit.bindings.listener1-in-0.consumer.exchangeType=topic
spring.cloud.stream.rabbit.bindings.listener1-in-0.consumer.autoBindDlq=true

# Listener 2
spring.cloud.stream.bindings.listener2-in-0.destination=exchange2
spring.cloud.stream.bindings.listener2-in-0.group=exchange2-queue
spring.cloud.stream.rabbit.bindings.listener2-in-0.consumer.queueNameGroupOnly=true
spring.cloud.stream.rabbit.bindings.listener2-in-0.consumer.binding-routing-key-delimiter=,
spring.cloud.stream.rabbit.bindings.listener2-in-0.consumer.bindingRoutingKey=binding.key.2,binding.key.2.1
spring.cloud.stream.rabbit.bindings.listener2-in-0.consumer.exchangeType=topic
spring.cloud.stream.rabbit.bindings.listener2-in-0.consumer.autoBindDlq=true

build.gradle.kts：

import org.jetbrains.kotlin.gradle.tasks.KotlinCompile

plugins {
    id("org.springframework.boot") version "2.4.3"
    id("io.spring.dependency-management") version "1.0.11.RELEASE"
    kotlin("jvm") version "1.4.30"
    kotlin("plugin.spring") version "1.4.30"
}

group = "com.example"
version = "0.0.1-SNAPSHOT"
java.sourceCompatibility = JavaVersion.VERSION_1_8

repositories {
    mavenCentral()
}

extra["springCloudVersion"] = "2020.0.1"

dependencies {
    implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
    implementation("org.springframework.boot:spring-boot-starter-webflux")
    implementation("org.jetbrains.kotlin:kotlin-reflect")
    implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
    implementation("org.springframework.cloud:spring-cloud-stream")
    implementation("org.springframework.cloud:spring-cloud-stream-binder-rabbit")
    testImplementation("org.springframework.boot:spring-boot-starter-test")
}

dependencyManagement {
    imports {
        mavenBom("org.springframework.cloud:spring-cloud-dependencies:${property("springCloudVersion")}")
    }
}

tasks.withType<KotlinCompile> {
    kotlinOptions {
        freeCompilerArgs = listOf("-Xjsr305=strict")
        jvmTarget = "1.8"
    }
}

tasks.withType<Test> {
    useJUnitPlatform()
}

当我注释掉一个侦听器bean时，另一个侦听器bean按预期工作。但是，如果这两个bean都处于活动状态，则不会在RabbitMQ中创建队列，如果我手动创建队列并向交换发送消息，也不会从队列中读取消息。

我在这里做错什么了？

Answer 1

该框架只能检测单个函数。当您有多个时，您需要指定：

spring.cloud.function.definition=listener1;listener2

https://docs.spring.io/spring-cloud-stream/docs/3.1.1/reference/html/spring-cloud-stream.html#spring_cloud_function

在只有java.util.function.[Supplier/Function/Consumer]类型的单个bean的情况下，可以跳过spring.cloud.function.definition属性，因为这样的函数bean将被自动发现。但是，使用此类属性以避免任何混淆被认为是最佳实践。