我正在嘗試為使用 RabbitMQ 的應用程序編寫集成測試，為此我正在使用 Qpid 代理。我設法啟動了服務器并且我的測試正在連接到它，但我需要在啟動前在 Qpid 中創建隊列。因為我有大量隊列，所以我動態創建了 bean：applicationContext.getBeanFactory().registerSingleton(queueName, queue);這需要在啟動前創建隊列。這是 qpid 配置文件：{ "name": "tst", "modelVersion": "2.0", "defaultVirtualHost" : "default", "authenticationproviders" : [ { "name" : "noPassword", "type" : "Anonymous", "secureOnlyMechanisms": [] }, { "name" : "passwordFile", "type" : "PlainPasswordFile", "path" : "/src/test/resources/passwd.txt", "secureOnlyMechanisms": [], "preferencesproviders" : [{ "name": "fileSystemPreferences", "type": "FileSystemPreferences", "path" : "${qpid.work_dir}${file.separator}user.preferences.json" } ] } ], "ports" : [ { "name": "AMQP", "port": "5673", "authenticationProvider": "passwordFile", "protocols": [ "AMQP_0_10", "AMQP_0_8", "AMQP_0_9", "AMQP_0_9_1" ] }], "virtualhostnodes" : [ { "name" : "default", "type" : "JSON", "virtualHostInitialConfiguration" : "{ \"type\" : \"Memory\" }" }]}從官方文檔（https://qpid.apache.org/releases/qpid-broker-j-7.1.4/book/Java-Broker-Management-Channel-REST-API.html#d0e2130）我讀到可以為 REST 調用創建隊列，所以我嘗試使用 RestTemplate 來實現這一點，但它似乎沒有創建隊列。 @BeforeClass public static void startup() throws Exception { brokerStarter = new BrokerManager(); brokerStarter.startBroker(); RestTemplate restTemplate = new RestTemplate(); restTemplate.put("http://localhost:5673/api/latest/queue/default/queue1", ""); restTemplate.put("http://localhost:5673/api/latest/queue/default/queue-2", ""); }有人可以解釋我做錯了什么嗎？謝謝你！
2 回答
人到中年有點甜
TA貢獻1895條經驗 獲得超7個贊
我使用 REST API 解決了同樣的問題。為了創建/刪除用于集成測試目的的隊列，我使用以下配置文件 ( qpid-config.json)：
{
"name": "EmbeddedBroker",
"modelVersion": "8.0",
"authenticationproviders": [
{
"name": "anonymous",
"type": "Anonymous"
}
],
"ports": [
{
"name": "AMQP",
"bindingAddress": "localhost",
"port": "${qpid.amqp_port}",
"protocols": [ "AMQP_1_0" ],
"authenticationProvider": "anonymous",
"virtualhostaliases" : [ {
"name" : "nameAlias",
"type" : "nameAlias"
}, {
"name" : "defaultAlias",
"type" : "defaultAlias"
}, {
"name" : "hostnameAlias",
"type" : "hostnameAlias"
} ]
},
{
"name" : "HTTP",
"port" : "${qpid.http_port}",
"protocols" : [ "HTTP" ],
"authenticationProvider" : "anonymous"
}
],
"virtualhostnodes": [
{
"name": "default",
"defaultVirtualHostNode": "true",
"type": "Memory",
"virtualHostInitialConfiguration": "{\"type\": \"Memory\" }"
}
],
"plugins" : [
{
"type" : "MANAGEMENT-HTTP",
"name" : "httpManagement"
}
]
}
相關的 Gradle 依賴項：
testImplementation("org.apache.qpid:qpid-broker-core:${Versions.qpidBroker}") // tested with 8.0.0
testImplementation("org.apache.qpid:qpid-broker-plugins-amqp-1-0-protocol:${Versions.qpidBroker}")
testImplementation("org.apache.qpid:qpid-broker-plugins-memory-store:${Versions.qpidBroker}")
testImplementation("org.apache.qpid:qpid-broker-plugins-management-http:${Versions.qpidBroker}")
testImplementation("org.springframework.boot:spring-boot-starter-webflux")
testImplementation("org.projectreactor:reactor-spring:${Versions.reactorSpring}")
啟動代理的代碼（Kotlin）：
private fun startQpidBroker() {
val attributes: MutableMap<String, Any> = HashMap()
val initialConfig = EmbeddedAMQPBroker::class.java.classLoader.getResource("qpid-config.json")!!
attributes["type"] = "Memory"
attributes["initialConfigurationLocation"] = initialConfig.toExternalForm()
attributes["startupLoggedToSystemOut"] = true
System.setProperty("qpid.amqp_port", "$amqpPort")
System.setProperty("qpid.http_port", "$httpPort")
// needed to avoid "AMQP precondition failed" due to durable message being sent to non-durable queues
System.setProperty("qpid.tests.mms.messagestore.persistence", "true")
broker.startup(attributes)
}
刪除/創建隊列的代碼：
private fun recreateQueue(queueName: String) {
val client = WebClient.create("http://localhost:${EmbeddedAMQPBroker.httpPort}");
try {
client.method(HttpMethod.DELETE)
.uri("/api/latest/queue/default/$queueName")
.retrieve()
.toBodilessEntity()
.block()
.statusCode
} catch (e: WebClientResponseException) {
if (e.statusCode != HttpStatus.NOT_FOUND) { // queue might not yet exist so 404 is acceptable
throw e
}
}
client.method(HttpMethod.PUT)
.uri("/api/latest/queue/default/default/$queueName")
.header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
.body(BodyInserters.fromValue(mapOf("name" to queueName, "type" to "standard")))
.retrieve()
.toBodilessEntity()
.block()
.statusCode
}
至尊寶的傳說
TA貢獻1789條經驗 獲得超10個贊
我設法通過使用連接工廠解決了這個問題：
@Autowired
ConnectionFactory factory;
....
factory.setHost("localhost");
factory.setPort(qpid_server_port);
try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
String queue = "queue-x";
channel.queueDeclare(queue, true, false, false, null);
//channel.queueBind(queue, "exchange-x" , "routing-key-x");
} catch (Exception e) {
e.printStackTrace();
}
添加回答
舉報
0/150
提交
取消