Add BatchingProcessor #1
Conversation
| int capacity; | ||
| long maxRetryCount; | ||
|
|
||
| private final ConcurrentMap<String, BufferedTaskGroup> windowedTasks = new ConcurrentHashMap<>(64, 0.75f, |
There was a problem hiding this comment.
capacityやload factorなどを詳細に設定する理由ってなんでしょうか？
There was a problem hiding this comment.
すみません、参考元をそのままコピペしてしまっただけで特に理由はないですmm
| @Override | ||
| public void process(ProcessingContext<T> context, T task) throws InterruptedException { | ||
| String key = context.key(); | ||
| BufferedTaskGroup taskGroup = windowedTasks.get(key); |
There was a problem hiding this comment.
見たところkeyごとにbatchingする設計のようですが、key単位でのバッチングは以下のようなデメリットがあるかと思います。
- keyのcardinalityが高い時に、バッチングが効かずにほとんどlingerでのflushとなる
key単位でのバッチングが有効なケースもあるとは思いますが、Decatonでbuilt-inで提供することを考えると、シンプルにBatchingProcessor instance単位でのバッチングが行われる、というほうがより分かりやすいし多くのユースケースにマッチするのではないでしょうか。
There was a problem hiding this comment.
instance 単位でバッチングが行えるように書き換えました。(大部分の変更になってしまいレビュー負担をおかけしてすみません)
| } | ||
|
|
||
| BatchingProcessor( | ||
| BiConsumer<String, List<BufferedTask<T>>> processor, |
There was a problem hiding this comment.
実際のprocessingロジックをBiConsumerで受け取るのではなく、BatchingProcessorをabstract classにしてuserに実装してもらうほうが、processorのlifecycle管理をDecatonに移譲できてよいと思いました。
|
https://github.com/line/decaton 側に draft PR を直接たてるため、この PR はクローズします |
進捗確認用 PR