使用重新连接逻辑连接到 Neptune 数据库实例的 Java 示例 - Amazon Neptune
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

使用重新连接逻辑连接到 Neptune 数据库实例的 Java 示例

以下 Java 示例演示了如何使用重新连接逻辑连接到 Gremlin 客户端,以便从意外的断开连接中恢复。

它具有以下依赖项:

<dependency> <groupId>org.apache.tinkerpop</groupId> <artifactId>gremlin-driver</artifactId> <version>${gremlin.version}</version> </dependency> <dependency> <groupId>com.amazonaws</groupId> <artifactId>amazon-neptune-sigv4-signer</artifactId> <version>${sig4.signer.version}</version> </dependency> <dependency> <groupId>com.amazonaws</groupId> <artifactId>amazon-neptune-gremlin-java-sigv4</artifactId> <version>${sig4.signer.version}</version> </dependency> <dependency> <groupId>com.evanlennick</groupId> <artifactId>retry4j</artifactId> <version>0.15.0</version> </dependency>

以下是示例代码:

public static void main(String args[]) { boolean useIam = true; // Create Gremlin cluster and traversal source Cluster.Builder builder = Cluster.build() .addContactPoint(System.getenv("neptuneEndpoint")) .port(Integer.parseInt(System.getenv("neptunePort"))) .enableSsl(true) .minConnectionPoolSize(1) .maxConnectionPoolSize(1) .serializer(Serializers.GRAPHBINARY_V1D0) .reconnectInterval(2000); if (useIam) { builder = builder.channelizer(SigV4WebSocketChannelizer.class); } Cluster cluster = builder.create(); GraphTraversalSource g = AnonymousTraversalSource .traversal() .withRemote(DriverRemoteConnection.using(cluster)); // Configure retries RetryConfig retryConfig = new RetryConfigBuilder() .retryOnCustomExceptionLogic(getRetryLogic()) .withDelayBetweenTries(1000, ChronoUnit.MILLIS) .withMaxNumberOfTries(5) .withFixedBackoff() .build(); @SuppressWarnings("unchecked") CallExecutor<Object> retryExecutor = new CallExecutorBuilder<Object>() .config(retryConfig) .build(); // Do lots of queries for (int i = 0; i < 100; i++){ String id = String.valueOf(i); @SuppressWarnings("unchecked") Callable<Object> query = () -> g.V(id) .fold() .coalesce( unfold(), addV("Person").property(T.id, id)) .id().next(); // Retry query // If there are connection failures, the Java Gremlin client will automatically // attempt to reconnect in the background, so all we have to do is wait and retry. Status<Object> status = retryExecutor.execute(query); System.out.println(status.getResult().toString()); } cluster.close(); } private static Function<Exception, Boolean> getRetryLogic() { return e -> { Class<? extends Exception> exceptionClass = e.getClass(); StringWriter stringWriter = new StringWriter(); String message = stringWriter.toString(); if (RemoteConnectionException.class.isAssignableFrom(exceptionClass)){ System.out.println("Retrying because RemoteConnectionException"); return true; } // Check for connection issues if (message.contains("Timed out while waiting for an available host") || message.contains("Timed-out waiting for connection on Host") || message.contains("Connection to server is no longer active") || message.contains("Connection reset by peer") || message.contains("SSLEngine closed already") || message.contains("Pool is shutdown") || message.contains("ExtendedClosedChannelException") || message.contains("Broken pipe") || message.contains(System.getenv("neptuneEndpoint"))) { System.out.println("Retrying because connection issue"); return true; }; // Concurrent writes can sometimes trigger a ConcurrentModificationException. // In these circumstances you may want to backoff and retry. if (message.contains("ConcurrentModificationException")) { System.out.println("Retrying because ConcurrentModificationException"); return true; } // If the primary fails over to a new instance, existing connections to the old primary will // throw a ReadOnlyViolationException. You may want to back and retry. if (message.contains("ReadOnlyViolationException")) { System.out.println("Retrying because ReadOnlyViolationException"); return true; } System.out.println("Not a retriable error"); return false; }; }