Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

在 Java 中开发 Kinesis 客户端库用户

可以使用 Kinesis Client Library (KCL) 构建处理 Kinesis 数据流中数据的应用程序。Kinesis Client Library 有多种语言版本。本主题将讨论 Java。要查看 Javadoc 参考，请参阅Amazon班级的 Javadoc 主题AmazonKinesis客户端.

从下载 Java KCLGitHub，转到Kinesis 客户端库 (Java). 要在 Apache Maven 上找到 Java KCL，请转至KCL 搜索结果页. 从下载 Java KCL 消费者应用程序的示例代码GitHub，转到KCL for Java 示例项目上的页面GitHub.

该示例应用程序使用 Apache Commons Logging。可以使用 configure 文件中定义的静态 AmazonKinesisApplicationSample.java 方法更改日志记录配置。有关如何将 Apache Commons 日志记录与 Log4j 结合使用的更多信息和AmazonJava 应用程序，请参阅使用 Log4j 日志记录中的Amazon SDK for Java开发人员指南.

在 Java 中实现 KCL 使用者应用程序时，您必须完成下列任务：

实施 IRecordProcessor方法

KCL 目前支持两个版本IRecordProcessor接口：原始接口与 KCL 的第一个版本一起可用；而版本 2 从 KCL 1.5.0 版开始可用。这两个接口都完全受支持。您的选择取决于您的特定方案要求。要查看所有区别，请参阅您在本地构建的 Javadocs 或源代码。以下各节概述了开始使用的最低实施要求。

原始接口（版本 1）

原始 IRecordProcessor 接口 (package com.amazonaws.services.kinesis.clientlibrary.interfaces) 公开了下列记录处理器方法，您的使用者必须实施这些方法。该示例提供了可用作起点的实现（请参阅 AmazonKinesisApplicationSampleRecordProcessor.java）。

public void initialize(String shardId)
public void processRecords(List<Record> records, IRecordProcessorCheckpointer checkpointer)
public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason) 

initialize

KCL 称initialize方法在实例化记录处理器时，将特定分片 ID 作为参数传递。此记录处理器仅处理此分片，并且通常情况下反过来说也成立（此分片仅由此记录处理器处理）。但是，您的使用者应该考虑数据记录可能会经过多次处理的情况。Kinesis Data Streams 有至少一次语义，即分片中的每个数据记录至少会由使用者中的工作程序处理一次。有关特定分片可能由多个工作程序进行处理的情况的更多信息，请参阅重新分片、扩展和并行处理。

public void initialize(String shardId)

processRecords

KCL 称processRecords方法，并传递来自由指定的分片的数据记录的列表。initialize(shardId)方法。记录处理器根据使用者的语义处理这些记录中的数据。例如，工作程序可能执行数据转换，然后将结果存储在 Amazon Simple Storage Service (Amazon S3) 存储桶中。

public void processRecords(List<Record> records, IRecordProcessorCheckpointer checkpointer) 

除了数据本身之外，记录还包含一个序号和一个分区键。工作程序可在处理数据时使用这些值。例如，工作线程可选择 S3 存储桶，并在其中根据分区键的值存储数据。Record 类公开了以下方法，这些方法提供对记录的数据、序列号和分区键的访问。

record.getData()  
record.getSequenceNumber() 
record.getPartitionKey() 

在该示例中，私有方法 processRecordsWithRetries 具有显示工作程序如何访问记录的数据、序号和分区键的代码。

Kinesis Data Streams 需要记录处理器来跟踪已在分片中处理的记录。KCL 通过传递检查指针来为您执行此跟踪（IRecordProcessorCheckpointer) 到processRecords. 记录处理器调用checkpoint方法以向 KCL 通知记录处理分片中的记录的进度。如果工作线程失败，KCL 将使用此信息在已知的上一个已处理记录处重新启动对分片的处理。

对于拆分或合并操作，在原始分片的处理器调用之前，KCL 不会开始处理新分片。checkpoint表示原始分片的所有处理都已完成。

如果你没有传递参数，KCL 假定调用checkpoint表示所有记录都已处理，一直处理到传递到记录处理器的最后一个记录。因此，记录处理器只应在已处理传递到它的列表中的所有记录后才调用 checkpoint。记录处理器不需要在每次调用 checkpoint 时调用 processRecords。例如，处理器可以在每第三次调用 checkpoint 时调用 processRecords。您可以选择性地将某个记录的确切序号指定为 checkpoint 的参数。在本例中，KCL 将假定所有记录都已处理，一直到达该记录。

在该示例中，私有方法 checkpoint 展示了如何使用适当的异常处理和重试逻辑调用 IRecordProcessorCheckpointer.checkpoint。

KCL 依靠processRecords以处理由处理数据记录引起的任何异常。如果从中抛出异常processRecords，KCL 将跳过在异常发生前已传递的数据记录。也就是说，这些记录不会重新发送到引发异常的记录处理器或使用者中的任何其他记录处理器。

shutdown

KCL 称shutdown方法无论是在处理结束时（关闭原因是TERMINATE) 或工作程序不再响应（关闭原因为ZOMBIE)。

public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason)

处理操作在记录处理器不再从分片中接收任何记录时结束，因为分片已被拆分或合并，或者流已删除。

KCL 还通过了IRecordProcessorCheckpointer接口到shutdown. 如果关闭原因为 TERMINATE，则记录处理器应完成处理任何数据记录，然后对此接口调用 checkpoint 方法。

更新后的接口（版本 2）

更新后的 IRecordProcessor 接口 (package com.amazonaws.services.kinesis.clientlibrary.interfaces.v2) 公开了下列记录处理器方法，您的使用者必须实施这些方法：

void initialize(InitializationInput initializationInput)
void processRecords(ProcessRecordsInput processRecordsInput)
void shutdown(ShutdownInput shutdownInput)

原始版本的接口中的所有参数可通过容器对象上的 get 方法进行访问。例如，要检索 processRecords() 中的记录的列表，可使用 processRecordsInput.getRecords()。

自此接口的版本 2（KCL 1.5.0 及更高版本）起，除了原始接口提供的输入之外，以下新输入也可用：

为 I 实现类工厂RecordProcessor接口

您还需要为实现记录处理器方法的类实现一个工厂。当使用者实例化工作程序时，它将传递对此工厂的引用。

以下示例使用原始记录处理器接口在文件 AmazonKinesisApplicationSampleRecordProcessorFactory.java 中实现工厂类。如果您希望此工厂类创建版本 2 记录处理器，请使用程序包名称 com.amazonaws.services.kinesis.clientlibrary.interfaces.v2。

  public class SampleRecordProcessorFactory implements IRecordProcessorFactory { 
      /**
      * Constructor.
      */
      public SampleRecordProcessorFactory() {
          super();
      }
      /**
      * {@inheritDoc}
      */
      @Override
      public IRecordProcessor createProcessor() {
          return new SampleRecordProcessor();
      }
  } 

创建工作线程

正如中讨论的实施 IRecordProcessor方法，有两个版本的 KCL 记录处理器接口可供选择，这将影响您创建工作程序的方式。原始记录处理器接口使用以下代码结构创建工作线程：

final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...)
final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
final Worker worker = new Worker(recordProcessorFactory, config);

当使用版本 2 的记录处理器接口时，您可使用 Worker.Builder 创建工作线程而无需担心要使用的构造函数以及参数的顺序。更新后的记录处理器接口使用以下代码结构创建工作线程：

final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...)
final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
final Worker worker = new Worker.Builder()
    .recordProcessorFactory(recordProcessorFactory)
    .config(config)
    .build();

修改配置属性

该示例提供了配置属性的默认值。工作程序的此配置数据随后将整合到 KinesisClientLibConfiguration 对象中。此对象和对 IRecordProcessor 的类工厂的引用将传入用于实例化工作程序的调用。您可借助 Java 属性文件（请参阅 AmazonKinesisApplicationSample.java）用您自己的值覆盖任何这些属性。

Application Name

KCL 需要一个应用程序名称，该名称在您的应用程序中以及同一区域的 Amazon DynamoDB 表中是唯一的。KCL 通过以下方法使用应用程序名称配置值：

设置凭证

你必须让你Amazon凭证提供程序链中的凭证提供程序之一提供给凭证。例如，如果您正在 EC2 实例上运行您的使用者，我们建议您使用 IAM 角色启动实例。Amazon反映与此 IAM 角色关联的权限的凭证通过实例的元数据提供给实例上的应用程序。使用这种方式管理在 EC2 实例上运行的使用者的凭证最安全。

示例应用程序首先尝试从实例元数据中检索 IAM 凭证：

credentialsProvider = new InstanceProfileCredentialsProvider(); 

如果示例应用程序无法从实例元数据中获取凭证，它会尝试从属性文件中检索凭证：

credentialsProvider = new ClasspathPropertiesFileCredentialsProvider();

有关实例元数据的更多信息，请参阅实例元数据中的适用于 Linux 实例的 Amazon EC2 用户指南.

将工作程序 ID 用于多个实例

示例初始化代码通过使用本地计算机的名称并附加一个全局唯一的标识符为工作程序创建 ID (workerId)，如以下代码段所示。此方法支持使用者应用程序的多个实例在单台计算机上运行的方案。

String workerId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + UUID.randomUUID();

迁移到版本 2 的记录处理器接口

如果要迁移使用原始接口的代码，则除了上述步骤之外，还需要执行以下步骤：