Simple Log Service：Gunakan Aliyun Log Java Producer untuk menulis data log ke Layanan Log Sederhana

Future

Metode send mengembalikan ListenableFuture, yang selain metode get standar juga memungkinkan pendaftaran callback setelah selesai. Kode sampel di bawah ini menunjukkan penggunaan ListenableFuture. Daftarkan FutureCallback untuk future dan jalankan di kolam thread EXECUTOR_SERVICE yang disediakan aplikasi. Untuk contoh lengkap, lihat SampleProducerWithFuture.java.

package com.aliyun.openservices.aliyun.log.producer.sample;

import com.aliyun.openservices.aliyun.log.producer.*;
import com.aliyun.openservices.aliyun.log.producer.errors.LogSizeTooLargeException;
import com.aliyun.openservices.aliyun.log.producer.errors.MaxBatchCountExceedException;
import com.aliyun.openservices.aliyun.log.producer.errors.ProducerException;
import com.aliyun.openservices.aliyun.log.producer.errors.ResultFailedException;
import com.aliyun.openservices.aliyun.log.producer.errors.TimeoutException;
import com.aliyun.openservices.log.common.LogItem;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SampleProducerWithFuture {

  private static final Logger LOGGER = LoggerFactory.getLogger(SampleProducerWithFuture.class);

  private static final ExecutorService EXECUTOR_SERVICE = Executors
      .newFixedThreadPool(Math.max(Runtime.getRuntime().availableProcessors(), 1));

  public static void main(String[] args) throws InterruptedException {
    Producer producer = Utils.createProducer();
    int n = 100;
    int size = 20;

    // Jumlah log yang telah selesai (baik berhasil dikirim maupun gagal)
    final AtomicLong completed = new AtomicLong(0);

    for (int i = 0; i < n; ++i) {
      List<LogItem> logItems = Utils.generateLogItems(size);
      try {
        String project = System.getenv("PROJECT");
        String logStore = System.getenv("LOG_STORE");
        ListenableFuture<Result> f = producer.send(project, logStore, logItems);
        Futures.addCallback(
            f, new SampleFutureCallback(project, logStore, logItems, completed), EXECUTOR_SERVICE);
      } catch (InterruptedException e) {
        LOGGER.warn("Thread saat ini telah terganggu selama pengiriman log.");
      } catch (Exception e) {
        if (e instanceof MaxBatchCountExceedException) {
          LOGGER.error("Log melebihi jumlah batch maksimum, e={}", e);
        } else if (e instanceof LogSizeTooLargeException) {
          LOGGER.error("Ukuran log lebih besar dari ukuran maksimum yang diizinkan, e={}", e);
        } else if (e instanceof TimeoutException) {
          LOGGER.error("Waktu yang diperlukan untuk mengalokasikan memori untuk log telah terlampaui., e={}", e);
        } else {
          LOGGER.error("Gagal mengirim log, e=", e);
        }
      }
    }

    Utils.doSomething();

    try {
      producer.close();
    } catch (InterruptedException e) {
      LOGGER.warn("Thread saat ini telah terganggu dari penutupan.");
    } catch (ProducerException e) {
      LOGGER.info("Gagal menutup producer, e=", e);
    }

    EXECUTOR_SERVICE.shutdown();
    while (!EXECUTOR_SERVICE.isTerminated()) {
      EXECUTOR_SERVICE.awaitTermination(100, TimeUnit.MILLISECONDS);
    }
    LOGGER.info("Semua log selesai, completed={}", completed.get());
  }

  private static final class SampleFutureCallback implements FutureCallback<Result> {

    private static final Logger LOGGER = LoggerFactory.getLogger(SampleFutureCallback.class);

    private final String project;

    private final String logStore;

    private final List<LogItem> logItems;

    private final AtomicLong completed;

    SampleFutureCallback(
        String project, String logStore, List<LogItem> logItems, AtomicLong completed) {
      this.project = project;
      this.logStore = logStore;
      this.logItems = logItems;
      this.completed = completed;
    }

    @Override
    public void onSuccess(@Nullable Result result) {
      LOGGER.info("Kirim log berhasil.");
      completed.getAndIncrement();
    }

    @Override
    public void onFailure(Throwable t) {
      if (t instanceof ResultFailedException) {
        Result result = ((ResultFailedException) t).getResult();
        LOGGER.error(
            "Gagal mengirim log, project={}, logStore={}, result={}", project, logStore, result);
      } else {
        LOGGER.error("Gagal mengirim log, e=", t);
      }
      completed.getAndIncrement();
    }
  }
}

Callback

Callback dieksekusi oleh thread internal producer, dan ruang data hanya dilepaskan setelah selesai. Untuk mencegah pemblokiran producer dan mengurangi throughput, hindari operasi panjang dalam callback. Selain itu, jangan panggil metode send untuk percobaan ulang dalam callback. Sebagai gantinya, tangani percobaan ulang dalam callback ListenableFuture. Untuk contoh lengkap, lihat SampleProducerWithCallback.java.

package com.aliyun.openservices.aliyun.log.producer.sample;

import com.aliyun.openservices.aliyun.log.producer.Callback;
import com.aliyun.openservices.aliyun.log.producer.Producer;
import com.aliyun.openservices.aliyun.log.producer.Result;
import com.aliyun.openservices.aliyun.log.producer.errors.LogSizeTooLargeException;
import com.aliyun.openservices.aliyun.log.producer.errors.ProducerException;
import com.aliyun.openservices.aliyun.log.producer.errors.TimeoutException;
import com.aliyun.openservices.log.common.LogItem;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SampleProducerWithCallback {

  private static final Logger LOGGER = LoggerFactory.getLogger(SampleProducerWithCallback.class);

  private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(10);

  public static void main(String[] args) throws InterruptedException {
    final Producer producer = Utils.createProducer();

    int nTask = 100;

    // Nomor urut monotonik yang akan dimasukkan dalam data setiap log
    final AtomicLong sequenceNumber = new AtomicLong(0);

    // Jumlah log yang telah selesai (baik berhasil dikirim maupun gagal)
    final AtomicLong completed = new AtomicLong(0);

    final CountDownLatch latch = new CountDownLatch(nTask);

    for (int i = 0; i < nTask; ++i) {
      EXECUTOR_SERVICE.submit(
          new Runnable() {
            @Override
            public void run() {
              LogItem logItem = Utils.generateLogItem(sequenceNumber.getAndIncrement());
              try {
                String project = System.getenv("PROJECT");
                String logStore = System.getenv("LOG_STORE");
                producer.send(
                    project,
                    logStore,
                    Utils.getTopic(),
                    Utils.getSource(),
                    logItem,
                    new SampleCallback(project, logStore, logItem, completed));
              } catch (InterruptedException e) {
                LOGGER.warn("Thread saat ini telah terganggu selama pengiriman log.");
              } catch (Exception e) {
                if (e instanceof LogSizeTooLargeException) {
                  LOGGER.error(
                      "Ukuran log lebih besar dari ukuran maksimum yang diizinkan, e={}", e);
                } else if (e instanceof TimeoutException) {
                  LOGGER.error(
                      "Waktu yang diperlukan untuk mengalokasikan memori untuk log telah terlampaui., e={}", e);
                } else {
                  LOGGER.error("Gagal mengirim log, logItem={}, e=", logItem, e);
                }
              } finally {
                latch.countDown();
              }
            }
          });
    }
    latch.await();
    EXECUTOR_SERVICE.shutdown();

    Utils.doSomething();

    try {
      producer.close();
    } catch (InterruptedException e) {
      LOGGER.warn("Thread saat ini telah terganggu dari penutupan.");
    } catch (ProducerException e) {
      LOGGER.info("Gagal menutup producer, e=", e);
    }

    LOGGER.info("Semua log selesai, completed={}", completed.get());
  }

  private static final class SampleCallback implements Callback {

    private static final Logger LOGGER = LoggerFactory.getLogger(SampleCallback.class);

    private final String project;

    private final String logStore;

    private final LogItem logItem;

    private final AtomicLong completed;

    SampleCallback(String project, String logStore, LogItem logItem, AtomicLong completed) {
      this.project = project;
      this.logStore = logStore;
      this.logItem = logItem;
      this.completed = completed;
    }

    @Override
    public void onCompletion(Result result) {
      try {
        if (result.isSuccessful()) {
          LOGGER.info("Kirim log berhasil.");
        } else {
          LOGGER.error(
              "Gagal mengirim log, project={}, logStore={}, logItem={}, result={}",
              project,
              logStore,
              logItem.ToJsonString(),
              result);
        }
      } finally {
        completed.getAndIncrement();
      }
    }
  }
}

Shutdown aman

Shutdown aman direkomendasikan dalam kebanyakan kasus, menggunakan metode close(). Metode close() menunggu semua data yang disimpan cache diproses, thread berhenti, callback dieksekusi, dan future selesai sebelum kembali.

Meskipun metode ini menunggu semua data diproses, ia segera kembali jika callback tidak diblokir, dan batch segera diproses tanpa percobaan ulang setelah penutupan.

Shutdown terbatas

Untuk pengembalian cepat ketika callback mungkin memblokir, gunakan shutdown terbatas dengan metode close(long timeoutMs). Jika producer tidak sepenuhnya ditutup setelah timeoutMs yang ditentukan, IllegalStateException akan dilemparkan, menunjukkan potensi kehilangan data dan callback yang belum dieksekusi.