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

Future

Metode send mengembalikan ListenableFuture, yang selain metode get standar juga memungkinkan pendaftaran callback setelah selesai. Contoh kode 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 diinterupsi 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 dibutuhkan 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 diinterupsi saat menutup.");
    } 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("Berhasil mengirim log.");
      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 penurunan throughput, hindari operasi yang memakan waktu lama di dalam callback. Selain itu, jangan panggil metode send untuk retry di dalam callback. Sebaliknya, tangani retry di 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 ke 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 diinterupsi 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 dibutuhkan 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 diinterupsi saat menutup.");
    } 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("Berhasil mengirim log.");
        } else {
          LOGGER.error(
              "Gagal mengirim log, project={}, logStore={}, logItem={}, result={}",
              project,
              logStore,
              logItem.ToJsonString(),
              result);
        }
      } finally {
        completed.getAndIncrement();
      }
    }
  }
}

Shutdown aman

Shutdown aman direkomendasikan dalam sebagian besar kasus, menggunakan metode close(). Metode close() menunggu semua data yang di-cache diproses, thread dihentikan, callback dieksekusi, dan future diselesaikan sebelum mengembalikan hasil.

Meskipun metode ini menunggu semua data diproses, metode ini cepat mengembalikan hasil jika callback tidak diblokir, dan batch segera diproses tanpa retry setelah penutupan.

Shutdown terbatas

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