同一Kinesis流的多个不同消费者

Question

我有一个Kinesis生产者，它将单一类型的消息写入流。我想在多个完全不同的消费者应用程序中处理这个流。因此，给定主题/流的具有单个发布者的发布/订阅。我还想利用检查点来确保每个消费者处理写入流的每条消息。

最初，我对所有消费者和生产者使用相同的应用程序名称。但是，一旦我启动多个消费者，我就开始收到以下错误：

通用域名格式。amazonaws。服务。运动。模型InvalidArgumentException:StartingSequenceNumber 4956423629634456659779527257172303925768853369405442在账户*******下创建的流包中的shard shardId-000000000000上的GetShardIterator中使用，因为它不是来自此流，所以无效。（服务：AmazonKinesis；状态代码：400；错误代码：InvalidArgumentException；请求ID：…）

这似乎是因为消费者在使用相同的应用程序名称时与他们的检查点发生冲突。

从阅读文档来看，使用检查点进行发布/订阅的唯一方法似乎是拥有一个针对每个消费者的流应用程序，这要求每个生产者了解所有可能的消费者。这比我想要的更紧密地结合在一起；这真的只是一个队列。

Kafka似乎支持我想要的：任意使用给定的主题/分区，因为消费者完全可以控制自己的检查点。如果我想要有检查点的酒吧/酒吧，我唯一的选择是搬到Kafka还是其他的选择？

我的RecordProcessor代码，在每个消费者中都是相同的：

override def processRecords(processRecordsInput: ProcessRecordsInput): Unit = {
  log.trace("Received record(s) from kinesis")
  for {
    record <- processRecordsInput.getRecords
    json   <- jawn.parseByteBuffer(record.getData).toOption
    msg    <- decode[T](json.toString).toOption
  } yield subscriber ! msg
  processRecordsInput.getCheckpointer.checkpoint()
}

代码解析消息并将其发送给订阅者。目前，我只是将所有消息标记为已成功接收。我可以在AWS Kinesis仪表板上看到正在发送的消息，但不会发生读取，可能是因为每个应用程序都有自己的AppName，并且看不到任何其他消息。

Answer 1

您想要的模式，一个发布者的模式

你是怎么做到的？您需要为每个消费者指定不同的应用程序名称。这样，一个消费者的检查点信息就不会与另一个消费者的检查点信息发生冲突。

检查对此的第一个响应：https://forums.aws.amazon.com/message.jspa?messageID=554375