Google Cloud PubSub将消息发送给多个用户（在同一订阅中）

Question

private final static int ACK_DEAD_LINE_IN_SECONDS = 30;
private static final long POLLING_PERIOD_MS = 250L;
private static final int WINDOW_MAX_SIZE = 1000;
private static final Duration WINDOW_MAX_TIME = Duration.ofSeconds(1L);

@Autowired
private PubSubAdmin pubSubAdmin;

@Bean
public ApplicationRunner runner(PubSubReactiveFactory reactiveFactory) {
    return args -> {
        createSubscription("subscription-id", "topic-id", ACK_DEAD_LINE_IN_SECONDS);
        reactiveFactory.poll(subscription, POLLING_PERIOD_MS) // Poll the PubSub periodically
            .map(msg -> Pair.of(msg, getMessageValue(msg))) // Extract the message as a pair
            .bufferTimeout(WINDOW_MAX_SIZE, WINDOW_MAX_TIME) // Create a buffer of messages to bulk process 
            .flatMap(this::processBuffer) // Process the buffer
            .doOnError(e -> log.error("Error processing event window", e))
            .retry()
            .subscribe();
    };
}

private void createSubscription(String subscriptionName, String topicName, int ackDeadline) {
    pubSubAdmin.createTopic(topicName);
    try {
        pubSubAdmin.createSubscription(subscriptionName, topicName, ackDeadline);
    } catch (AlreadyExistsException e) {
        log.info("Pubsub subscription '{}' already configured for topic '{}': {}", subscriptionName, topicName, e.getMessage());
    }
}

private Flux<Void> processBuffer(List<Pair<AcknowledgeablePubsubMessage, PreparedRecordEvent>> msgsWindow) {
    return Flux.fromStream(
        msgsWindow.stream()
            .collect(Collectors.groupingBy(msg -> msg.getRight().getData())) // Group the messages by same data
            .values()
            .stream()
    )
    .flatMap(this::processDataBuffer);
}

private Mono<Void> processDataBuffer(List<Pair<AcknowledgeablePubsubMessage, PreparedRecordEvent>> dataMsgsWindow) {
    return processData(
        dataMsgsWindow.get(0).getRight().getData(),
        dataMsgsWindow.stream()
            .map(Pair::getRight)
            .map(PreparedRecordEvent::getRecord)
            .collect(Collectors.toSet())
    )
    .doOnSuccess(it ->
        dataMsgsWindow.forEach(msg -> {
            log.info("Mark msg ACK");
            msg.getLeft().ack();
        })
    )
    .doOnError(e -> {
        log.error("Error on PreparedRecordEvent event", e);
        dataMsgsWindow.forEach(msg -> {
            log.error("Mark msg NACK");
            msg.getLeft().nack();
        });
    })
    .retry();
}

private Mono<Void> processData(Data data, Set<Record> records) {
    // For each message, make calculations over the records associated to the data
    final DataQuality calculated = calculatorService.calculateDataQualityFor(data, records); // Arithmetic calculations
    return this.daasClient.updateMetrics(calculated) // Update DB record with a DaaS to wrap DB access
        .flatMap(it -> {
            if (it.getProcessedRows() >= it.getValidRows()) {
                return finish(data);
            }
            return Mono.just(data);
        })
        .then();
}

private Mono<Data> finish(Data data) {
    return dataClient.updateStatus(data.getId, DataStatus.DONE) // Update DB record with a DaaS to wrap DB access
        .doOnSuccess(updatedData -> pubSubClient.publish(
            new Qa0DonedataEvent(updatedData) // Publis a new event in other topic
        ))
        .doOnError(err -> {
            log.error("Error finishing data");
        })
        .onErrorReturn(data);
}

根据官方文档，一旦消息被发送给订阅者，pub/sub就会尝试不将其传递给同一订阅的任何其他订阅者(app2实例是同一订阅的订阅者）：

一旦消息被发送到订阅服务器，订阅服务器应该确认该消息。消息一旦发送出去并在订阅者确认之前被视为未完成。pub/sub将反复尝试传递任何未被确认的消息。然而，当一条消息对订阅者来说是未完成的时，Pub/Sub会尝试不将它传递给同一订阅中的任何其他订阅者。订阅服务器有一个可配置的、有限的时间--称为ackDeadline--来确认未完成的消息。一旦截止日期过去，消息不再被认为是未完成的，并且pub/sub将尝试重新传递消息