So maximieren Sie den Durchsatz mit RequestResponseo auf dem GCP -Datenfluss ⇐ Java

[Anonymous]

Ich versuche, RequestResponseo auf DataFlow zu verwenden, um parallele Anforderungen an einen Endpunkt zu stellen. Als Test habe ich einen Cloud Run HelloWorld -Endpunkt erstellt, der diese Anfragen nur empfängt. Es kann bis zu 80 gleichzeitige Anfragen pro Instanz bearbeiten und standardmäßig bis zu 100 Instanzen skalieren. Die Anforderungslatenz beträgt im Allgemeinen ~ 1 ms, aber rrio maxes bei ~ 100 Elementen/s.

Code: Select all

public class MakeRequest implements Caller {
private static HttpRequestFactory REQUEST_FACTORY;
private static final Logger LOG = LoggerFactory.getLogger(MakeRequest.class);
private static final String url = "REDACTED";

public MakeRequest() {}

public HttpRequestFactory getRequestFactory() {...}

public static IdTokenCredentials generateIdTokenCredentials(String url) throws IOException {...}

public static HttpRequestFactory createHttpFactory(String url) {...}

public String call(String reqString) {
HttpRequest request;
try {
request = getRequestFactory().buildGetRequest(new GenericUrl(url));
} catch (IOException e) {
throw new RuntimeException(e);
}

HttpResponse response = null;
try {
response = request.execute();
return response.getContent().toString();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}

Es gibt viele zusätzliche Code, um die Authentifizierung mit Cloud -Lauf zu verarbeiten, die ich ausgeschnitten habe, aber der Großteil der Logik befindet sich in der Aufruf -Funktion , die ich den Dokumenten hier erstellt habe. Da es sich nur um Testcode handelt, lasse ich die übergebenen Stringelemente fallen. Run gibt nicht einmal einen 429 keine verfügbaren Instanzfehler zurück, daher ist er eindeutig nicht überladen. Ich habe versucht, eine benutzerdefinierte Callshouldbackoff -implementierung hinzuzufügen, die sich anmelden sollte, wenn die Funktionen update () aufgerufen werden, und es scheint überhaupt nicht zu treffen:
.apply(
RequestResponseIO.of(new MakeRequest(), StringUtf8Coder.of())
.withCallShouldBackoff(new TestCallShouldBackoffBasedOnRejectionProbability()));
< /code>
Fehlt mir eine Möglichkeit, die Parallelität oder den Durchsatz der Funktion zu erhöhen? Wie stellt Rrio fest, dass es die Rate verlangsamen sollte, mit der es die API nennt? Mit 1000 Elementen im Körperlauf dauert es ~ 1s, die Anforderung zu verarbeiten, wobei 100 Elemente im Körperlauf ~ 0,1s benötigen, um die Anforderung zu verarbeiten. Rrio scheint diese mit 1/s oder 10/s zu betreiben, was zu impliziert, dass es eine Art Kappe trifft, wenn der Lauf absolut viel höher skaliert wird.
tia!