Cree una suscripción

Función de suscripción

Cuando consume datos de un tema de DataHub, debe gestionar sus propios offsets de consumo para reanudar el procesamiento tras un error de la aplicación. Esto requiere guardar su progreso y asegurarse de que su servicio de almacenamiento de offsets tenga alta disponibilidad, lo que añade complejidad a la aplicación. Para simplificarlo, DataHub proporciona un servicio de suscripción que almacena sus offsets de consumo en el servidor. Con unos pasos sencillos de configuración y un código mínimo, obtiene un servicio de gestión de offsets de alta disponibilidad que funciona de forma transparente para su aplicación. Además, ofrece capacidades flexibles de restablecimiento de offsets, compatibles con la semántica de consumo de al menos una vez. Por ejemplo, si detecta un error de procesamiento que afectó a datos de un periodo específico y desea volver a consumirlos, puede restablecer el offset al momento correspondiente. Su aplicación detecta automáticamente este cambio y vuelve a procesar los datos sin necesidad de reiniciar.

Asegúrese de que su cuenta tenga permiso para crear una suscripción a un tema del proyecto especificado. Para obtener más información, consulte la documentación de control de permisos. Siga estos pasos:

Abra la página Topic, haga clic en + Subscription en la esquina superior derecha, complete los detalles de la suscripción y haga clic en Create.
- Subscription Application: Nombre de la aplicación que usa esta suscripción.
- Description: Descripción detallada de la suscripción.
Haga clic en el botón de búsqueda situado debajo de Consumption Checkpoint para consultar el estado de consumo de todos los shards.

Ejemplo de uso

La función de suscripción almacena offsets. Es independiente de las funciones de lectura y escritura de DataHub (consulte la documentación del SDK de Java), pero suele usarse con ellas cuando necesita almacenar offsets de consumo después de leer datos.

// Example of consuming data and committing offsets during the process.
public void offset_consumption(int maxRetry) {
    String endpoint = "<YourEndPoint>";
    String accessId = "<YourAccessId>";
    String accessKey = "<YourAccessKey>";
    String projectName = "<YourProjectName>";
    String topicName = "<YourTopicName>";
    String subId = "<YourSubId>";
    String shardId = "0";
    List<String> shardIds = Arrays.asList(shardId);
    // Create a DatahubClient instance.
    DatahubClient datahubClient = DatahubClientBuilder.newBuilder()
            .setDatahubConfig(
                    new DatahubConfig(endpoint,
                            // Whether to enable binary transfer. This feature is supported by the server since version 2.12.
                            new AliyunAccount(accessId, accessKey), true))
            .build();
    RecordSchema schema = datahubClient.getTopic(projectName, topicName).getRecordSchema();
    OpenSubscriptionSessionResult openSubscriptionSessionResult = datahubClient.openSubscriptionSession(projectName, topicName, subId, shardIds);
    SubscriptionOffset subscriptionOffset = openSubscriptionSessionResult.getOffsets().get(shardId);
    // 1. Get the cursor for the current offset. If the current offset has expired or has never been consumed, get the cursor for the first record within the lifecycle.
    String cursor = "";
    // A sequence number less than 0 indicates that the shard has not been consumed.
    if (subscriptionOffset.getSequence() < 0) {
        // Get the cursor for the first record within the lifecycle.
        cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.OLDEST).getCursor();
    } else {
        // Get the cursor for the next record.
        long nextSequence = subscriptionOffset.getSequence() + 1;
        try {
            // Getting a cursor using SEQUENCE may throw a SeekOutOfRangeException, which indicates that the data at the current cursor has expired.
            cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.SEQUENCE, nextSequence).getCursor();
        } catch (SeekOutOfRangeException e) {
            // Get the cursor for the first record within the lifecycle.
            cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.OLDEST).getCursor();
        }
    }
    // 2. Read records and save the offset. This example demonstrates reading tuple data and committing the offset every 1,000 records.
    long recordCount = 0L;
    // Read 1,000 records at a time.
    int fetchNum = 1000;
    int retryNum = 0;
    int commitNum = 1000;
    while (retryNum < maxRetry) {
        try {
            GetRecordsResult getRecordsResult = datahubClient.getRecords(projectName, topicName, shardId, schema, cursor, fetchNum);
            if (getRecordsResult.getRecordCount() <= 0) {
                // No data. Sleep and try again.
                System.out.println("no data, sleep 1 second");
                Thread.sleep(1000);
                continue;
            }
            for (RecordEntry recordEntry : getRecordsResult.getRecords()) {
                // Process the data.
                TupleRecordData data = (TupleRecordData) recordEntry.getRecordData();
                System.out.println("field1:" + data.getField("field1") + "\t"
                        + "field2:" + data.getField("field2"));
                // After processing the data, update the offset.
                recordCount++;
                subscriptionOffset.setSequence(recordEntry.getSequence());
                subscriptionOffset.setTimestamp(recordEntry.getSystemTime());
                // Commit the offset every 1000 records.
                if (recordCount % commitNum == 0) {
                    // Commit the offset.
                    Map<String, SubscriptionOffset> offsetMap = new HashMap<>();
                    offsetMap.put(shardId, subscriptionOffset);
                    datahubClient.commitSubscriptionOffset(projectName, topicName, subId, offsetMap);
                    System.out.println("commit offset successful");
                }
            }
            cursor = getRecordsResult.getNextCursor();
        } catch (SubscriptionOfflineException | SubscriptionSessionInvalidException e) {
            // Exit. SubscriptionOfflineException: The subscription is offline. SubscriptionSessionInvalidException: Another client is consuming the same subscription.
            e.printStackTrace();
            throw e;
        } catch (SubscriptionOffsetResetException e) {
            // The offset was reset. You need to get the latest version of the SubscriptionOffset.
            SubscriptionOffset offset = datahubClient.getSubscriptionOffset(projectName, topicName, subId, shardIds).getOffsets().get(shardId);
            subscriptionOffset.setVersionId(offset.getVersionId());
            // After an offset is reset, you must get the new cursor. The method you use to get the cursor should match how the offset was reset.
            // If both sequence and timestamp were set during the reset, you can get the cursor by using either SEQUENCE or SYSTEM_TIME.
            // If only the sequence was set, you must use SEQUENCE.
            // If only the timestamp was set, you must use SYSTEM_TIME.
            // As a general rule, try to get the cursor using SEQUENCE first, then SYSTEM_TIME. If both fail, use OLDEST.
            cursor = null;
            if (cursor == null) {
                try {
                    long nextSequence = offset.getSequence() + 1;
                    cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.SEQUENCE, nextSequence).getCursor();
                    System.out.println("get cursor successful");
                } catch (DatahubClientException exception) {
                    System.out.println("get cursor by SEQUENCE failed, try to get cursor by SYSTEM_TIME");
                }
            }
            if (cursor == null) {
                try {
                    cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.SYSTEM_TIME, offset.getTimestamp()).getCursor();
                    System.out.println("get cursor successful");
                } catch (DatahubClientException exception) {
                    System.out.println("get cursor by SYSTEM_TIME failed, try to get cursor by OLDEST");
                }
            }
            if (cursor == null) {
                try {
                    cursor = datahubClient.getCursor(projectName, topicName, shardId, CursorType.OLDEST).getCursor();
                    System.out.println("get cursor successful");
                } catch (DatahubClientException exception) {
                    System.out.println("get cursor by OLDEST failed");
                    System.out.println("get cursor failed!!");
                    throw e;
                }
            }
        } catch (LimitExceededException e) {
            // Limit exceeded, retry.
            e.printStackTrace();
            retryNum++;
        } catch (DatahubClientException e) {
            // Other error, retry.
            e.printStackTrace();
            retryNum++;
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(-1);
        }
    }
}

Cuando la aplicación se inicia por primera vez, comienza a consumir datos desde el registro disponible más antiguo. A medida que la aplicación se ejecuta, actualice la página de la suscripción en la consola web para consultar el avance del offset de consumo del shard.
Si cambia manualmente el offset mediante la función Reset Checkpoint de la consola web mientras el consumidor se está ejecutando, la aplicación detecta automáticamente el cambio y reanuda el consumo desde el nuevo offset. Para ello, el cliente captura una excepción SubscriptionOffsetResetException y llama al método getSubscriptionOffset para recuperar el objeto SubscriptionOffset más reciente del servidor.
No use varios hilos o procesos de consumidor para consumir simultáneamente el mismo shard de una suscripción. Esto provoca que diferentes consumidores sobrescriban el offset, lo que deja el offset almacenado en un estado indefinido. En este escenario, el servidor lanza una excepción SubscriptionSessionInvalidException. Capture esta excepción, salga de la aplicación y compruebe en su diseño si hay consumidores duplicados.