通过订阅从Azure服务总线接收消息

Question

我有一个超时选项，只想在超时前接收消息。

如果您能解释下面的代码是如何工作的，以及我如何修改下面的代码以在特定的时间框架内接收消息，并且一旦我的超时已经到达就停止接收，这将是很有帮助的。

static 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) {
                    // receives message is passed to callback
                    if (message.getLabel() != null &&
                            message.getContentType() != null &&
                            message.getLabel().contentEquals("Scientist") &&
                            message.getContentType().contentEquals("application/json")) {

                        byte[] body = message.getBody();
                        Map scientist = GSON.fromJson(new String(body, UTF_8), Map.class);

                        System.out.printf(
                                "\n\t\t\t\t%s Message received: \n\t\t\t\t\t\tMessageId = %s, \n\t\t\t\t\t\tSequenceNumber = %s, \n\t\t\t\t\t\tEnqueuedTimeUtc = %s," +
                                        "\n\t\t\t\t\t\tExpiresAtUtc = %s, \n\t\t\t\t\t\tContentType = \"%s\",  \n\t\t\t\t\t\tContent: [ firstName = %s, name = %s ]\n",
                                receiveClient.getEntityPath(),
                                message.getMessageId(),
                                message.getSequenceNumber(),
                                message.getEnqueuedTimeUtc(),
                                message.getExpiresAtUtc(),
                                message.getContentType(),
                                scientist != null ? scientist.get("firstName") : "",
                                scientist != null ? scientist.get("name") : "");
                    }
                    return receiveClient.completeAsync(message.getLockToken());
                }

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

Answer 1

在您的订阅代码中无法做到这一点（发布-订阅的整个思想不是这样工作的）。

有两种选择/变通方法可供选择：

• 以后不要向主题发送消息。
• 创建使REST API调用订阅的计时器触发器-创建或更新以使EntityStatus=ReceiveDisabled。