Tablestore SDK for Java, bir tünel üzerinden verileri sürekli olarak tüketmenize, her toplu işi bir geri arama ile işlemeye ve kalp atışlarını, denetim noktalarını, iş parçacığı havuzları ile tüketim eşzamanlılığını yapılandırmanıza olanak tanır.
Kullanım notları
Artımlı günlüklerin saklama süresi, tablonun Stream günlük sona erme süresiyle aynıdır ve en fazla yedi gün olabilir. BaseAndStream türündeki bir tünel için tam veri tüketimi bu süre içinde tamamlanmazsa, artımlı veri tüketimi başladığında
OTSTunnelExpiredhatası döndürülür. Tünel, artımlı veri tüketimine devam edemez.Artımlı tüketim saklama süresinin gerisinde kalırsa, tünel en son kullanılabilir veriden devam edebilir. Bu durumda bazı veriler tüketilmeyebilir.
Süresi dolmuş bir tünel devre dışı bırakılabilir. Tünel 30 günden fazla süreyle devre dışı kalırsa silinir ve geri yüklenemez.
Ön koşullar
Tablestore SDK for Java öğesini yükleyin ve TunnelClient'ı başlatın.
Özellik açıklaması
TunnelWorker öğesi, tünel kimliği kullanarak bir tünele bağlanır, kalp atışları aracılığıyla mevcut istemciye atanan kanalları alır, verileri sürekli çeker ve her kayıt topluluğunu IChannelProcessor öğesine iletir. Birden çok TunnelWorker örneği aynı tüneli tüketirse, sunucu kanalları istemciler arasında dağıtır.
Bir tünel üzerinden veri tüketmek için:
IChannelProcessoruygulayın. Her toplu işi işlemek içinprocessyöntemini ve geri aramanın kullandığı kaynakları serbest bırakmak içinshutdownyöntemini kullanın.Geri aramayı ve tüketim davranışını yapılandırmak üzere bir
TunnelWorkerConfignesnesi oluşturun.Tünel kimliği,
TunnelClientveTunnelWorkerConfigile birTunnelWorkeroluşturun.-
Tüketimi başlatmak için
connectAndWorkingçağrısını yapın.void process(ProcessRecordsInput input); void shutdown();
Aşağıdaki örnek, tünel üzerinden çekilen her kaydı yazdırır ve ardından tüketimi başlatır.
private static class SimpleProcessor implements IChannelProcessor {
@Override
public void process(ProcessRecordsInput input) {
for (StreamRecord record : input.getRecords()) {
System.out.println(record);
}
}
@Override
public void shutdown() {
// Release resources used by the callback.
}
}
String tunnelId = "example_tunnel_id";
TunnelWorkerConfig config =
new TunnelWorkerConfig(new SimpleProcessor());
TunnelWorker worker =
new TunnelWorker(tunnelId, tunnelClient, config);
worker.connectAndWorking();connectAndWorking arka plan tüketim görevlerini başlattıktan sonra döner. Uygulama sürecinin çalışır durumda kalmasını sağlayın. Tüketimi durdurmak için worker.shutdown(), config.shutdown() ve tunnelClient.shutdown() çağrılarını bu sırayla yapın. worker.shutdown() tünel bağlantılarını kapatır ve geri aramanın shutdown yöntemini çağırır. config.shutdown() okuma, işleme ve yardımcı iş parçacığı havuzlarını kapatır. TunnelWorker çalışanı durdurmaya çalışan bir JVM kapatma hook'u kaydeder ancak uygulamanın bu kaynakları açıkça serbest bırakması gerekir.
Parametreler
Çalışan
TunnelWorker oluşturucusu aşağıdaki parametreleri içerir.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Tüketim yapılandırması
workerConfig öğesi TunnelWorkerConfig türündedir ve aşağıdaki parametreleri içerir.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Aynı makinede birden çok TunnelWorker örneği başlatırsanız tek bir TunnelWorkerConfig nesnesini yeniden kullanarak okuma ve işleme iş parçacığı havuzlarını paylaşabilirsiniz. Tüm çalışanlar durduktan sonra config.shutdown() çağrısını yalnızca bir kez yapın.
Geri arama verileri
process yöntemi aşağıdaki alanları içeren bir ProcessRecordsInput nesnesi alır.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Senaryo örnekleri
Tüketim parametrelerini ayarlama
Tüketim aktarım hızı veya bellek kullanımı gereksinimlerinizi karşılamıyorsa kalp atışı ve denetim noktası aralıklarını, kanal eşzamanlılığını, tur başına çekme sayısı ve boyutunu ile artımlı çekmeler için geri çekilme aralığını ayarlayın.
TunnelWorkerConfig config =
new TunnelWorkerConfig(new SimpleProcessor());
config.setHeartbeatIntervalInSec(10);
config.setHeartbeatTimeoutInSec(60);
config.setCheckpointIntervalInMillis(10_000);
config.setMaxChannelParallel(16);
config.setReadMaxTimesPerRound(4);
config.setReadMaxBytesPerRound(8 * 1024 * 1024);
config.setMaxRetryIntervalInMillis(3_000);