使用 apache qpid 使用 amqpwss 连接到 azure 服务总线端口 443

Question

我正在尝试使用 amqp-10-jms-spring-boot 启动器连接到 Azure 服务总线（我知道在后台使用 apache qpid jms。

当我将连接字符串 (amqphub.amqp10jms.remote-url) 设置为使用“amqps://”时，一切正常，hello world 消息被发送到 Azure 队列，然后由应用程序检索和打印。

但是，当我使用“amqpwss://[Endpoint]:443”连接字符串时，我收到一个异常...

Caused by: io.netty.handler.codec.http.websocketx.WebSocketHandshakeException: Invalid handshake response getStatus: 400 This service does not support WebSocket connections.

我怀疑 Azure 服务总线不支持 WebSocket 连接。

1. amqphub 文档声称这是可能的：https://github.com/amqphub/amqp-10-jms-spring-boot#jms-connection-configuration
2. Apache quid 文档也是如此：http://qpid.apache.org/releases/qpid-jms-0.52.0/docs/index.html
3. 与服务总线文档一样“AMQP WebSockets 绑定在 TCP 端口 443 上创建一个隧道，然后等效于 AMQP 5671 连接。”：https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-amqp-protocol-guide#connections-and-sessions

如何从 Spring Boot 应用程序建立到 Azure 服务总线的 amqpwss 连接？（我喜欢使用 qpid，但我不依赖它)。

此处的示例代码：https://github.com/kevvvvyp/amqp-websocket-apache-qpid

Answer 1

在尝试支持追加销售并再次查看文档后，我再次证明，微软非常热衷于接纳和供应商将人们锁定在他们的花园中。

除了 Tirade，连接到 websockets 端点的 URL 与常规 AMQP 端点不同。假设你有以下

amqps://[endpoint]

要使用 websockets 进行连接，您需要将协议替换为 amqpwss 并将端口替换为 443

amqpwss://[endpoint]:443

但是，没有提到的是，您还必须指定/$servicebus/websocket 的路径。结果，最终的url是

amqpwss://[endpoint]:443/$servicebus/websocket

我不知道你应该如何解决这个问题。请证明我错了，并指出截至 2021-03-18 确实包含此信息的文档中的位。无论如何，我是通过this SO question 发现的，关于通过 websockets 从浏览器连接到 azure 服务总线。

Answer 2

虽然我相信 Dragas 的答案是正确的解决方案，但我想如果它对任何人有用，我会添加我采用的方法。

我最终使用了 Java azure-service-bus 库，因为我发现它依赖于 qpid-proton-j-extensions，这是他们自己的 Apache Qpid 扩展库。存储库描述为“扩展 qpid-proton-j 库以通过 WEBSOCKETS 讨论 AMQP”......所以我认为这可以完成这项工作！

我使用non-descructive read创建了一个订阅客户端...

 /**
 * Connect & start listening to the azure service bus topic.
 */
public void start() {
    listeningTask = taskExecutor.submit(() -> {
        try {
            SubscriptionClient subscriptionClient = new SubscriptionClient(connectionString, ReceiveMode.PEEKLOCK);
            ExecutorService receiveExecutor = Executors.newCachedThreadPool();
            registerMessageHandlerOnClient(subscriptionClient, receiveExecutor);
        } catch (Exception e) {
            log.error("Caught exception", e);
        }
    });
}

订阅响应，通过网关将任何消息发送到我的 Spring Integration Flow。我添加了一些可配置的属性来在消息消费之间应用退避...

   /**
     * Azure service bus listener.
     *
     * @param receiveClient   client
     * @param executorService executorService
     * @throws Exception If we cannot poll queue.
     */
    private void registerMessageHandlerOnClient(SubscriptionClient receiveClient, ExecutorService executorService) throws Exception {
        // register the RegisterMessageHandler callback
        receiveClient.registerMessageHandler(
                new IMessageHandler() {
                    // callback invoked when the message handler loop has obtained a message
                    public CompletableFuture<Void> onMessageAsync(IMessage message) {

                        log.debug("Message received from azure, id: {}", message.getMessageId()); //TODO deprecation alternative
                        brokerGateway.send(message.getBody());

                        try {
                            Thread.sleep(backOff.toMillis());
                        } catch (InterruptedException e) {
                            log.error("Failed to apply azure backoff", e);
                        }
                        return CompletableFuture.completedFuture(null);
                    }

                    // callback invoked when the message handler has an exception to report
                    public void notifyException(Throwable throwable, ExceptionPhase exceptionPhase) {
                        log.error("Exception {}", exceptionPhase, throwable);
                    }
                },
                // 1 concurrent call, messages are auto-completed, auto-renew duration
                new MessageHandlerOptions(1, false, Duration.ofMinutes(1)),
                executorService);

    }