Tair (Redis® OSS-Compatible)：Implementasikan pemantauan granular halus dengan menggunakan TairTS

Ikhtisar TairTS

TairTS adalah modul internal dari Tair (Enterprise Edition) yang mendukung kueri dan penulisan real-time serta konkurensi tinggi. TairTS memungkinkan Anda memperbarui atau menambahkan data ke deret waktu yang ada, menggunakan algoritma kompresi gorilla dan penyimpanan spesifik untuk secara signifikan mengurangi biaya penyimpanan, serta menentukan pengaturan time-to-live (TTL) untuk skeys agar bergulir otomatis berdasarkan jendela waktu. Untuk informasi lebih lanjut, lihat TS.

Ikhtisar pemantauan granular halus

Gambar di atas menunjukkan arsitektur sistem pemantauan granular halus. Konsol mengirim konfigurasi pemantauan granular halus ke aplikasi, yang kemudian menulis konfigurasi tersebut ke kolektor menggunakan protokol MQ Telemetry Transport (MQTT). Kolektor memproses data konfigurasi dan menulis data ke database Tair.

• Kueri Konkurensi Tinggi

  Selama kueri konkurensi tinggi, TairTS memastikan performa kueri dan mendukung operasi agregat dalam skenario seperti downsampling, penyaringan berbasis atribut, kueri batch, serta penggunaan beberapa fungsi numerik untuk penyaringan dan kueri multi-level. Dengan TairTS, Anda dapat melakukan kueri batch dan agregasi menggunakan satu perintah untuk mengurangi interaksi jaringan, menerima respons dalam milidetik, dan mengidentifikasi masalah secara tepat waktu.

• Penulisan Konkurensi Tinggi

  Satu kolektor mungkin tidak cukup untuk menangani penulisan konkurensi tinggi seiring bertambahnya ukuran aplikasi. Untuk menyelesaikan masalah ini, TairTS memungkinkan Anda memperbarui atau menambahkan data ke deret waktu yang ada guna memastikan akurasi penulisan konkuren ke beberapa kolektor dan mengurangi penggunaan memori. Kode berikut memberikan contoh cara menulis data secara konkuren:

  import com.aliyun.tair.tairts.TairTs;
  import com.aliyun.tair.tairts.params.ExtsAggregationParams;
  import com.aliyun.tair.tairts.params.ExtsAttributesParams;
  import com.aliyun.tair.tairts.results.ExtsSkeyResult;
  import redis.clients.jedis.Jedis;
  
  public class test {
  
      protected static final String HOST = "127.0.0.1";
      protected static final int PORT = 6379;
  
  
      public static void main(String[] args) {
          try {
              Jedis jedis = new Jedis(HOST, PORT, 2000 * 100);
              if (!"PONG".equals(jedis.ping())) {
                  System.exit(-1);
              }
              TairTs tairTs = new TairTs(jedis);
              // Gunakan kode berikut jika Anda ingin bekerja dengan instance kluster:
              //TairTsCluster tairTsCluster = new TairTsCluster(jedisCluster);
  
              String pkey = "cpu_load";
              String skey1 = "app1";
              long startTs = (System.currentTimeMillis() - 100000) / 1000 * 1000;
              long endTs = System.currentTimeMillis() / 1000 * 1000;
              String startTsStr = String.valueOf(startTs);
              String endTsStr = String.valueOf(endTs);
  
              tairTs.extsdel(pkey, skey1);
              long num = 5;
  
              // Perbarui data secara konkuren di Collector A. 
              for (int i = 0; i < num; i++) {
                  double val = i;
                  long ts = startTs + i*1000;
                  String tsStr = String.valueOf(ts);
                  ExtsAttributesParams params = new ExtsAttributesParams();
                  params.dataEt(1000000000);
                  String addRet = tairTs.extsrawincr(pkey, skey1, tsStr, val, params);
              }
  
              ExtsAggregationParams paramsAgg = new ExtsAggregationParams();
              paramsAgg.maxCountSize(10);
              paramsAgg.aggAvg(1000);
  
              System.out.println("Hasil pembaruan Collector A:");
              ExtsSkeyResult rangeByteRet = tairTs.extsrange(pkey, skey1, startTsStr, endTsStr, paramsAgg);
              for (int i = 0; i < num; i++) {
                  System.out.println("    ts: " + rangeByteRet.getDataPoints().get(i).getTs() + ", value: " + rangeByteRet.getDataPoints().get(i).getDoubleValue());
              }
  
              // Perbarui data secara konkuren di Collector B. 
              for (int i = 0; i < num; i++) {
                  double val = i;
                  long ts = startTs + i*1000;
                  String tsStr = String.valueOf(ts);
                  ExtsAttributesParams params = new ExtsAttributesParams();
                  params.dataEt(1000000000);
                  String addRet = tairTs.extsrawincr(pkey, skey1, tsStr, val, params);
              }
  
  
              System.out.println("Hasil pembaruan Collector B:");
              rangeByteRet = tairTs.extsrange(pkey, skey1, startTsStr, endTsStr, paramsAgg);
              for (int i = 0; i < num; i++) {
                  System.out.println("    ts: " + rangeByteRet.getDataPoints().get(i).getTs() + ", value: " + rangeByteRet.getDataPoints().get(i).getDoubleValue());
              }
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  }

  Hasil:

  Hasil pembaruan Collector A:
      ts: 1597049266000, value: 0.0
      ts: 1597049267000, value: 1.0
      ts: 1597049268000, value: 2.0
      ts: 1597049269000, value: 3.0
      ts: 1597049270000, value: 4.0
  Hasil pembaruan Collector B:
      ts: 1597049266000, value: 0.0
      ts: 1597049267000, value: 2.0
      ts: 1597049268000, value: 4.0
      ts: 1597049269000, value: 6.0
      ts: 1597049270000, value: 8.0