需要会话的实体不可能为azure IMessageReceiver创建非会话消息接收器

Question

我正在尝试通过java应用程序从Azure服务总线获取消息。我创建了必要的客户端配置，例如，通过ManagementClient成功连接

@Bean
public ClientSettings getMessageReceiver() throws ServiceBusException, InterruptedException {

    AzureTokenCredentials azureTokenCredentials = new ApplicationTokenCredentials(
            "clientID,
            "domain",
            "secret",
            AzureEnvironment.AZURE
    );

    TokenProvider tokenProvider = TokenProvider.createAzureActiveDirectoryTokenProvider(
            new AzureAuthentication(azureTokenCredentials),
            AzureEnvironment.AZURE.activeDirectoryEndpoint(),
            null
    );

    ClientSettings clientSettings = new ClientSettings(tokenProvider,
            RetryPolicy.getDefault(),
            Duration.ofSeconds(30),
            TransportType.AMQP);

    return clientSettings;
}

ManagementClient managementClient =
            new ManagementClient(Util.convertNamespaceToEndPointURI("namespace"),
                    clientSettings);
    managementClient.getTopics();

但当我试图从特定主题获取消息时：

        SubscriptionClient subscriptionClient = new SubscriptionClient("namespace", "events/subscriptions/subscription", clientSettings, ReceiveMode.PEEKLOCK);

并得到一个错误消息：

需要会话的实体不可能创建非会话消息接收器。

应提供哪些附加步骤？

Answer 1

在主题订阅中创建时，您已启用会话（默认情况下已禁用）。如果不需要消息会话，请禁用“需要会话”重新创建订阅（注意：一旦创建订阅，您就不能更改该属性）。

或者，如果您确实需要消息会话，请像下面这样更新代码，以便首先接收会话，然后从接收的会话中接收消息。所有代码示例都可以在这里找到，会话示例也可以在这里找到。

java prettyprint-override">        // The connection string value can be obtained by:
        // 1. Going to your Service Bus namespace in Azure Portal.
        // 2. Go to "Shared access policies"
        // 3. Copy the connection string for the "RootManageSharedAccessKey" policy.
        String connectionString = "Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};"
            + "SharedAccessKey={key}";

        // Create a receiver.
        // "<<topic-name>>" will be the name of the Service Bus topic you created inside the Service Bus namespace.
        // "<<subscription-name>>" will be the name of the session-enabled subscription.
        ServiceBusReceiverAsyncClient receiver = new ServiceBusClientBuilder()
            .connectionString(connectionString)
            .sessionReceiver()
            .receiveMode(ReceiveMode.PEEK_LOCK)
            .topicName("<<topic-name>>")
            .subscriptionName("<<subscription-name>>")
            .buildAsyncClient();

        Disposable subscription = receiver.receiveMessages()
            .flatMap(context -> {
                if (context.hasError()) {
                    System.out.printf("An error occurred in session %s. Error: %s%n",
                        context.getSessionId(), context.getThrowable());
                    return Mono.empty();
                }

                System.out.println("Processing message from session: " + context.getSessionId());

                // Process message
                return receiver.complete(context.getMessage());
            }).subscribe(aVoid -> {
            }, error -> System.err.println("Error occurred: " + error));

        // Subscribe is not a blocking call so we sleep here so the program does not end.
        TimeUnit.SECONDS.sleep(60);

        // Disposing of the subscription will cancel the receive() operation.
        subscription.dispose();

        // Close the receiver.
        receiver.close();