Aliyun LOG Java Producer 快速入门

本文涉及的产品
对象存储 OSS,20GB 3个月
对象存储 OSS,内容安全 1000次 1年
对象存储 OSS,恶意文件检测 1000次 1年
简介: Aliyun LOG Java Producer 是为运行在大数据、高并发场景下的 Java 应用量身打造的高性能写 LogHub 类库。相对于原始的 API 或 SDK,使用 producer 写数据能为您带来诸多优势,包括高性能、计算与 I/O 逻辑分离、资源可控制等。

背景

Aliyun LOG Java Producer 是为运行在大数据、高并发场景下的 Java 应用量身打造的高性能写 LogHub 类库。相对于原始的 API 或 SDK,使用 producer 写数据能为您带来诸多优势,包括高性能、计算与 I/O 逻辑分离、资源可控制等。想要更深入地了解 producer 的功能和原理可参考文章日志上云利器 - Aliyun LOG Java Producer,本文将聚焦于 producer 的使用上。

使用步骤

使用 producer 可归结为如下图所示的三个步骤,下面分别为您介绍。
producer_quick_start

创建 producer

Producer 的创建过程主要涉及到以下对象。

ProjectConfig

ProjectConfig 包含目标 project 的服务入口信息以及表征调用者身份的访问凭证。

服务入口

最终访问地址会由 project 和 endpoint 拼出,关于如何确定 project 对应的 endpoint 可参考文章服务入口

访问凭证

Producer 支持用户配置 AK 或 STS token。如果使用 STS token,需要定期创建新的 ProjectConfig 然后将其 put 到 ProjectConfigs 里。

ProjectConfigs

如果您需要向不同的 project 中写入数据,可以创建多个 ProjectConfig 对象然后将它们 put 到 ProjectConfigs 里。ProjectConfigs 内部通过 map 维护不同 project 的配置,该 map 的 key 为 project,value 为该 project 对应的 Client。

ProducerConfig

ProducerConfig 用于配置发送策略,您可以根据不同的业务场景为参数指定不同的值,各参数含义如下表所示。

参数 类型 描述
totalSizeInBytes 整型 单个 producer 实例能缓存的日志大小上限,默认为 100MB。
maxBlockMs 整型 如果 producer 可用空间不足,调用者在 send 方法上的最大阻塞时间,默认为 60 秒。
如果超过这个时间后所需空间仍无法得到满足,send 方法会抛出 TimeoutException
如果将该值设为0,当所需空间无法得到满足时,send 方法会立即抛出 TimeoutException。
如果您希望 send 方法一直阻塞直到所需空间得到满足,可将该值设为负数。
ioThreadCount 整型 执行日志发送任务的线程池大小,默认为可用处理器个数。
batchSizeThresholdInBytes 整型 当一个 ProducerBatch 中缓存的日志大小大于等于 batchSizeThresholdInBytes 时,该 batch 将被发送,默认为 512 KB,最大可设置成 5MB。
batchCountThreshold 整型 当一个 ProducerBatch 中缓存的日志条数大于等于 batchCountThreshold 时,该 batch 将被发送,默认为 4096,最大可设置成 40960。
lingerMs 整型 一个 ProducerBatch 从创建到可发送的逗留时间,默认为 2 秒,最小可设置成 100 毫秒。
retries 整型 如果某个 ProducerBatch 首次发送失败,能够对其重试的次数,默认为 10 次。
如果 retries 小于等于 0,该 ProducerBatch 首次发送失败后将直接进入失败队列。
maxReservedAttempts 整型 每个 ProducerBatch 每次被尝试发送都对应着一个 Attempt,此参数用来控制返回给用户的 attempt 个数,默认只保留最近的 11 次 attempt 信息。
该参数越大能让您追溯更多的信息,但同时也会消耗更多的内存。
baseRetryBackoffMs 整型 首次重试的退避时间,默认为 100 毫秒。
Producer 采样指数退避算法,第 N 次重试的计划等待时间为 baseRetryBackoffMs * 2^(N-1)。
maxRetryBackoffMs 整型 重试的最大退避时间,默认为 50 秒。
adjustShardHash 布尔 如果调用 send 方法时指定了 shardHash,该参数用于控制是否需要对其进行调整,默认为 true。
buckets 整型 当且仅当 adjustShardHash 为 true 时,该参数才生效。此时,producer 会自动将 shardHash 重新分组,分组数量为 buckets。
如果两条数据的 shardHash 不同,它们是无法合并到一起发送的,会降低 producer 吞吐量。将 shardHash 重新分组后,能让数据有更多地机会被批量发送。
该参数的取值范围是 [1, 256],且必须是 2 的整数次幂,默认为 64。

LogProducer

LogProducer 是接口 Producer 的实现类,它接收唯一的参数 producerConfig。当您准备好 producerConfig 后,可以按照下列方式创建 producer 实例。

Producer producer = new LogProducer(producerConfig);

创建 producer 的同时会创建一系列线程,是一个相对昂贵的操作,因此建议一个应用共用一个 producer 实例。一个 producer 实例包含的线程如下表所示,其中 N 为该 producer 实例在当前进程中的编号,从 0 开始。

线程名格式 数量 描述
aliyun-log-producer-<N>-mover 1 负责将满足发送条件的 batch 投递到发送线程池里。
aliyun-log-producer-<N>-io-thread- ioThreadCount IOThreadPool 中真正用于执行数据发送任务的线程。
aliyun-log-producer-<N>-success-batch-handler 1 用于处理发送成功的 batch。
aliyun-log-producer-<N>-failure-batch-handler 1 用于处理发送失败的 batch。

另外,LogProducer 提供的所有方法都是线程安全的,可以在多线程环境下安全执行。

发送数据

Producer 实例创建好后,下一步就是使用其提供的方法发送数据。

参数介绍

Producer 接口提供多种发送方法,方法的各个参数含义如下。

参数 描述 能否为空
project 待发送数据的目标 project。 不能
logStore 待发送数据的目标 logStore。 不能
logItem 待发送数据。 不能
topic 待发送数据的 topic。
如果留空或没有指定,该字段将被赋予""。
source 待发送数据的 source。
如果留空或没有指定,该字段将被赋予 producer 所在宿主机的 IP。
shardHash 待发送数据的 shardHash,用于将数据写入 logStore 中的特定 shard。
如果留空或没有指定,数据将被随机写入目标 logStore 的某个 shard 中。
callback 用于告知调用者数据发送的结果。

另外,只有 project、logStore、topic、source、shardHash 这 5 个属性都相同的数据才有机会和并在一起批量发送。为了让数据合并功能充分发挥作用,同时也为了节省内存,建议您控制这 5 个字段的取值范围。如果某个字段如 topic 的取值确实非常多,建议您将其加入 logItem 而不是直接使用 topic。

获取发送结果

由于 producer 提供的所有发送方法都是异步的,需要通过返回的 future 或者传入的 callback 获取发送结果。

Future

Send 方法会返回一个 ListenableFuture,它除了可以像普通 future 那样通过调用 get 方法阻塞获得发送结果外,还允许你注册回调方法(回调方法会在完成 future 设置后被调用)。以下代码片段展示了 ListenableFuture 的使用方法,用户需要为该 future 注册一个 FutureCallback 并将其投递到应用提供的线程池 EXECUTOR_SERVICE 中执行,完整样例可参考 SampleProducerWithFuture.java

ListenableFuture<Result> f = producer.send("project", "logStore", logItem);
Futures.addCallback(f,
                    new FutureCallback<Result>() {
                        @Override
                        public void onSuccess(@Nullable Result result) {
                        }

                        @Override
                        public void onFailure(Throwable t) {
                        }
                    },
                    EXECUTOR_SERVICE);

Callback

除了使用 future 外,您还可以通过在调用 send 方法时注册 callback 获取数据发送结果,代码片段如下。(完整样例可参考 SampleProducerWithCallback.java

producer.send(
        "project",
        "logStore",
        logItem,
        new Callback() {
            @Override
            public void onCompletion(Result result) {
            }
        });

Callback 由 producer 内部线程负责执行,并且只有在执行完毕后数据“占用”的空间才会释放。为了不阻塞 producer 造成整体吞吐量的下降,要避免在 callback 里执行耗时的操作。另外,在 callback 中调用 send 方法进行重试也是不建议的,您可以上调参数 retries 或者在 ListenableFuture 的callback 中进行重试。

Future vs Callback

那么想要获取数据的发送结果到底是选择 future 还是 callback 呢?如果获取结果后,应用的处理逻辑比较简单且不会造成 producer 阻塞,建议直接使用 callback。否则,建议使用 ListenableFuture,在单独的线程(池)中执行后续业务。

关闭 producer

当您已经没有数据需要发送或者当前进程准备退出时,需要关闭 producer,目的是让 producer 中缓存的数据全部得到处理。目前,producer 提供安全关闭和有限关闭两种模式。

安全关闭

在大多数情况下,建议您使用安全关闭。安全关闭对应的方法是close(),它会等到 producer 中缓存的数据全部被处理、线程全部停止、注册的 callback 全部执行,返回 future 全部被设置后才会返回。

虽然要等到数据全部处理完成,但 producer 被关闭后,缓存的 batch 会被立刻处理且不会被重试。因此,如果 callback 不被阻塞,close 方法往往能在很短的时间内返回。

有限关闭

如果您的 callback 在执行过程中有可能阻塞,但您又希望 close 方法能在短时间内返回,可以使用有限关闭。有限关闭对应的方法是close(long timeoutMs),如果超过指定的 timeoutMs 后 producer 仍未完全关闭,它会抛出 IllegalStateException,这意味着缓存的数据可能还没来得及处理就被丢弃,用户注册的 callback 也可能不会被执行。

应用示例

为了进一步减少学习成本,我们为您准备了 Aliyun LOG Java Producer 应用示例。示例中包含了 producer 从创建到关闭的全部流程。

技术支持

日志服务-SLS

相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
相关文章
|
2月前
|
SQL Java 关系型数据库
【前端学java】JDBC快速入门
【8月更文挑战第12天】JDBC快速入门
23 2
【前端学java】JDBC快速入门
|
2月前
|
存储 前端开发 JavaScript
【前端学JAVA】有手就会!10min快速入门java的基础语法(2)
【8月更文挑战第8天】10min快速入门java的基础语法
33 2
【前端学JAVA】有手就会!10min快速入门java的基础语法(2)
|
2月前
|
消息中间件 Java Kafka
"Kafka快速上手:从环境搭建到Java Producer与Consumer实战,轻松掌握分布式流处理平台"
【8月更文挑战第10天】Apache Kafka作为分布式流处理平台的领头羊,凭借其高吞吐量、可扩展性和容错性,在大数据处理、实时日志收集及消息队列领域表现卓越。初学者需掌握Kafka基本概念与操作。Kafka的核心组件包括Producer(生产者)、Broker(服务器)和Consumer(消费者)。Producer发送消息到Topic,Broker负责存储与转发,Consumer则读取这些消息。首先确保已安装Java和Kafka,并启动服务。接着可通过命令行创建Topic,并使用提供的Java API实现Producer发送消息和Consumer读取消息的功能。
60 8
|
2月前
|
应用服务中间件 nginx Docker
[loki]轻量级日志聚合系统loki快速入门
[loki]轻量级日志聚合系统loki快速入门
|
2月前
|
消息中间件 Prometheus 监控
Producer的监控与日志记录最佳实践
【8月更文第29天】在分布式系统中,消息队列作为关键组件之一,其稳定性和性能至关重要。生产者(Producer)负责生成并发送消息到消息队列中,因此确保生产者的健康运行是非常重要的。本文将探讨如何为生产者设置监控和日志记录,以跟踪其健康状况和性能指标。
35 0
|
2月前
|
消息中间件 Java Kafka
【Azure 事件中心】开启 Apache Flink 制造者 Producer 示例代码中的日志输出 (连接 Azure Event Hub Kafka 终结点)
【Azure 事件中心】开启 Apache Flink 制造者 Producer 示例代码中的日志输出 (连接 Azure Event Hub Kafka 终结点)
|
3月前
|
Java 测试技术 Apache
《手把手教你》系列基础篇(八十六)-java+ selenium自动化测试-框架设计基础-Log4j实现日志输出(详解教程)
【7月更文挑战第4天】Apache Log4j 是一个广泛使用的 Java 日志框架,它允许开发者控制日志信息的输出目的地、格式和级别。Log4j 包含三个主要组件:Loggers(记录器)负责生成日志信息,Appenders(输出源)确定日志输出的位置(如控制台、文件、数据库等),而 Layouts(布局)则控制日志信息的格式。通过配置 Log4j,可以灵活地定制日志记录行为。
48 4
|
3月前
|
运维 Java Apache
Java中的日志框架:Log4j与SLF4J详解
Java中的日志框架:Log4j与SLF4J详解
|
3月前
|
XML Java 测试技术
《手把手教你》系列基础篇(八十八)-java+ selenium自动化测试-框架设计基础-Log4j 2实现日志输出-下篇(详解教程)
【7月更文挑战第6天】本文介绍了如何使用Log4j2将日志输出到文件中,重点在于配置文件的结构和作用。配置文件包含两个主要部分:`appenders`和`loggers`。`appenders`定义了日志输出的目标,如控制台(Console)或其他文件,如RollingFile,设置输出格式和策略。`loggers`定义了日志记录器,通过`name`属性关联到特定的类或包,并通过`appender-ref`引用`appenders`来指定输出位置。`additivity`属性控制是否继承父logger的配置。
37 0

热门文章

最新文章

相关产品

  • 日志服务