У меня есть Java-клиент google-pubsub api.
Я отправил 4 сообщения одновременно в паб.
Мой подписчик срабатывает один раз в час и зацикливается в пабе до тех пор, пока он не опустеет.
Тем не менее, я вижу, что сообщения извлекаются в разное время, хотя все они были отправлены в одно и то же время, и до следующего автоматического синхронизированного извлечения подписчика была большая задержка.
// A hotfix as cofman is too slow for ack
while (pubSubIsFull) {
// Fetch messages by order
log("before fetched messages by order");
List<ReceivedMessage> messages = mySyncSubscriber.fetch(100000, true);
messages = messages.stream().sorted(Comparator.comparingLong(message2 -> message2.getMessage().getPublishTime().getSeconds())).collect(Collectors.toList());
log("fetched " + messages.size() + " messages by order");
pubSubIsFull = messages.size() > 0;
log("before processMessages");
List<String> ackIds = processMessages(messages);
log("after processMessages");
mySyncSubscriber.sendAck(ackIds);
}
import com.google.pubsub.v1.PullRequest.Builder;
public List<ReceivedMessage> fetch(int maxMessages, boolean returnImmediately) {
String subscriptionName = this.getSubscriptionName(this.getSubscriptionId()).toString();
Builder pullRequestBuilder = PullRequest.newBuilder().setSubscription(subscriptionName).setReturnImmediately(returnImmediately);
if (maxMessages != 0) {
pullRequestBuilder.setMaxMessages(maxMessages);
}
PullRequest pullRequest = pullRequestBuilder.build();
PullResponse pullResponse = (PullResponse)this.subscriber.pullCallable().call(pullRequest);
return pullResponse.getReceivedMessagesList();
}
Это ошибка в пабе sub?
Есть ли что-то, что я могу сделать, чтобы изменить это?