Implementieren Sie den Verbraucher

Implementieren Sie die processRecord-Methode, indem Sie ein richtig bemessenes StockTrade-Objekt instanziieren und die Datensatzdaten zu diesem hinzufügen, sodass im Falle eines Problems eine Warnung protokolliert wird.

byte[] arr = new byte[record.data().remaining()];
record.data().get(arr);
StockTrade trade = StockTrade.fromJsonAsBytes(arr);
    if (trade == null) {
        log.warn("Skipping record. Unable to parse record into StockTrade. Partition Key: " + record.partitionKey());
        return;
        }
stockStats.addStockTrade(trade);
                

Implementieren Sie eine reportStats Methode. Ändern Sie das Ausgabeformat nach Ihren Wünschen.

System.out.println("****** Shard " + kinesisShardId + " stats for last 1 minute ******\n" +
stockStats + "\n" +
"****************************************************************\n");
                

Implementieren Sie die Methode resetStats, die eine neue stockStats-Instance erstellt.

stockStats = new StockStats();
                

Implementieren Sie die folgenden Methoden, die für die ShardRecordProcessor Schnittstelle erforderlich sind:

@Override
public void leaseLost(LeaseLostInput leaseLostInput) {
    log.info("Lost lease, so terminating.");
}

@Override
public void shardEnded(ShardEndedInput shardEndedInput) {
    try {
        log.info("Reached shard end checkpointing.");
        shardEndedInput.checkpointer().checkpoint();
    } catch (ShutdownException | InvalidStateException e) {
        log.error("Exception while checkpointing at shard end. Giving up.", e);
    }
}

@Override
public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
    log.info("Scheduler is shutting down, checkpointing.");
    checkpoint(shutdownRequestedInput.checkpointer());
}

private void checkpoint(RecordProcessorCheckpointer checkpointer) {
    log.info("Checkpointing shard " + kinesisShardId);
    try {
        checkpointer.checkpoint();
    } catch (ShutdownException se) {
        // Ignore checkpoint if the processor instance has been shutdown (fail over).
        log.info("Caught shutdown exception, skipping checkpoint.", se);
    } catch (ThrottlingException e) {
        // Skip checkpoint when throttled. In practice, consider a backoff and retry policy.
        log.error("Caught throttling exception, skipping checkpoint.", e);
    } catch (InvalidStateException e) {
        // This indicates an issue with the DynamoDB table (check for table, provisioned IOPS).
        log.error("Cannot save checkpoint to the DynamoDB table used by the Amazon Kinesis Client Library.", e);
    }
}
                

Führen Sie den unter Implementieren Sie den Hersteller erstellten Produzenten aus, um simulierte Wertpapiertransaktionsdatensätze in den Stream zu schreiben.

Stellen Sie sicher, dass der Zugriffsschlüssel und das geheime key pair, das Sie zuvor (beim Erstellen des IAM Benutzers) abgerufen haben, in der Datei gespeichert sind~/.aws/credentials.

Führen Sie die StockTradesProcessor-Klasse mit den folgenden Argumenten aus:

StockTradesProcessor StockTradeStream us-west-2

Beachten Sie, dass Sie, wenn Sie Ihren Stream in einer anderen Region als us-west-2 erstellt haben, stattdiesen diese Region hier angeben müssen.