我有一个关于flink和它的定时服务的问题。我有一个keyby流，它使用一个计时器，当计时器被调用时，我需要发送一个http请求，这可能需要时间来响应。我的问题是，我应该使http调用异步吗？或者flink已经将计时器调用作为一个新线程，每个键都是异步的？提前谢谢
goucqfw61#
如果你问 onTimer 方法是在一个单独的线程中为每个键调用的，那么我很确定它不是。因此，在这种情况下，您需要异步调用http调用。但说实话，我不认为这是一个好主意，一般来说，使用 onTimer 函数来执行http调用。我对您的用例一无所知，但是我认为您应该考虑使用不同的机制，比如创建边输出，然后使用flink Async 接线员。在内部使用异步调用 onTimer 这可能很棘手，尤其是当你考虑到重试、超时和可能的失败时。因此，根据评论，用例是每x分钟打一个电话为Kafka服务，然后把一些东西发到Kafka。所以，你可以做的是简单的有一个进程函数来调度计时器。每次启动计时器时，如果需要任何数据，则生成一些输出记录，其中包含请求所需的数据。然后你用 Async 运算符来实际执行请求，解析响应并基于响应生成一些输出记录，然后可以将这些记录保存到kafka。
onTimer
Async
zz2j4svz2#
你可以使用 ProcessFunction 它存储http请求所需的数据，并且可以具有计时器。当它触发时，您将发出一个包含请求数据的记录，该记录将由后续的 AsyncFunction 将用于提出您需要的定期请求。
ProcessFunction
AsyncFunction
2条答案
按热度按时间goucqfw61#
如果你问
onTimer方法是在一个单独的线程中为每个键调用的，那么我很确定它不是。因此，在这种情况下，您需要异步调用http调用。但说实话，我不认为这是一个好主意，一般来说，使用
onTimer函数来执行http调用。我对您的用例一无所知，但是我认为您应该考虑使用不同的机制，比如创建边输出，然后使用flinkAsync接线员。在内部使用异步调用onTimer这可能很棘手，尤其是当你考虑到重试、超时和可能的失败时。因此，根据评论，用例是每x分钟打一个电话为Kafka服务，然后把一些东西发到Kafka。所以，你可以做的是简单的有一个进程函数来调度计时器。每次启动计时器时，如果需要任何数据，则生成一些输出记录，其中包含请求所需的数据。然后你用
Async运算符来实际执行请求，解析响应并基于响应生成一些输出记录，然后可以将这些记录保存到kafka。zz2j4svz2#
你可以使用
ProcessFunction它存储http请求所需的数据，并且可以具有计时器。当它触发时，您将发出一个包含请求数据的记录，该记录将由后续的AsyncFunction将用于提出您需要的定期请求。