Apache NiFi——在Kafka消费者之后异步处理消息
目前我们使用Apache NiFi通过Kafka消费者消费消息。kafka消费者的输出连接到数据库处理器，数据库处理器从队列中获取消息（来自消费者）并在其上运行存储的过程/处理。因此，数据库处理器将处理来自队列的每条消息，我可以将数据库处理器设置为n个线程并行工作，但主要是每个线程可以处理每个队列的一条消息。
我期待做如下事情:
-
< li >消费者之后的处理器将仅消费队列中的消息(或获取消息),并说将等待“批处理”或总计1000条消息。 < li >一旦它收到1000条消息或60秒，并且消息计数为
所以我想知道任何一个这样的处理器，它可以执行上面的第2点，即，根据批处理/时间规则读取消息并将其推送到下一个？
共有1个答案
如果您可以利用NiFi的记录处理器，那么使用批量大小为1000的ConsumerKafkaRecord，然后使用PutDatabaseRecord，将提供与您描述的行为类似的行为。
如果在使用Kafka主题时，您可能总是没有足够的消息可用，那么在中间添加MergeContent或MergeRecord会让您等待一定的时间或一定数量的消息。
-
是否有一种方法以编程方式访问和打印使用者滞后偏移，或者说使用者读取的最后一条记录的偏移与某个生产者写入该使用者分区的最后一条记录的偏移之间的位置差。 要知道我的最终目标是将这个值发送到prometheus进行监视，我应该在上面添加哪些语句来得到滞后偏移值？
-
Flink kafka消费者有两种类型的消费者，例如： 这两个消费者层次结构扩展了相同的类。我想知道维护编号类背后的设计决策是什么？我们什么时候应该使用其中一种？ 我注意到带有数字后缀的类有更多的特性（例如ratelimiting）。 https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka
-
我有一个简单的Kafka设置。生成器正在以较高的速率向单个分区生成具有单个主题的消息。单个使用者正在使用来自此分区的消息。在此过程中，使用者可能会多次暂停处理消息。停顿可以持续几分钟。生产者停止产生消息后，所有排队的消息都将由使用者处理。生产者产生的消息似乎不会立即被消费者看到。我使用的是Kafka0.10.1.0。这里会发生什么？下面是使用消息的代码部分： 代理上的所有配置都保留为kafka默认
-
我刚接触Kafka，很少阅读教程。我无法理解使用者和分区之间的关系。 请回答我下面的问题。 > 消费者是否由ZK分配到单个分区，如果是，如果生产者将消息发送到不同的分区，那么其他分区的消费者将如何使用该消息？ 我有一个主题，它有3个分区。我发布消息，它会转到P0。我有5个消费者（不同的消费者群体）。所有消费者都会阅读P0的信息吗？若我增加了许多消费者，他们会从相同的P0中阅读信息吗？如果所有消费者
-
我在Flink的工作中使用Kafka资料来源的信息流，一次阅读50个主题，如下所示： 然后有一些运算符，如：过滤器- 我能获得的最大吞吐量是每秒10k到20k条记录，考虑到源发布了数十万个事件，这相当低，我可以清楚地看到消费者落后于生产者。我甚至试着移除水槽和其他操作员，以确保没有背压，但它仍然是一样的。我正在将我的应用程序部署到Amazon Kinesis data analytics，并尝试了
-
我目前正在探索Kafka，作为一个简单问题的初学者。 将有一个生产者向一个主题推送消息，但将有n个spark应用程序的消费者从kafka发送消息并插入到数据库中(每个消费者插入到不同的表中)。 是否有可能消费者会不同步（例如消费者的某些部分会停机很长一段时间），然后一个或多个消费者不会处理消息并插入到表中？ 假设代码总是正确的，在按摩数据时不会出现异常。重要的是每条消息只处理一次。 我的问题是，K
-
想知道Kafka使用者（Java客户端）是否可以并行读取和处理多条消息...我的意思是使用多个线程...我应该使用rxJava吗？？ 1）这样做是一个好的方法吗？？？2）而且根据我的理解，Kafka甚至把每一个线程都当作消费者...如果我错了，请纠正我... 3）并且还想让Java客户端作为守护进程服务在Linux中运行，这样它就可以连续运行，并且轮询Kafka的消息，读取和处理都是一样的...这
-
我正在使用这个库来实现节点kafka与消费者暂停和恢复方法来处理背压。我已经创建了一个小演示，我可以在其中和，但问题是在后它停止了消费消息。 这是我的代码。 任何人都可以帮助我，我在恢复消费者时做错了什么？当我启动使用者时，它只接收一条消息，并且在恢复后仍然不消耗任何其他消息。