leizhiyuan opened a new pull request, #1456: URL: https://github.com/apache/rocketmq-clients/pull/1456
### Which Issue(s) This PR Fixes Fixes #1447 ### Brief Description The synchronous `MessageListener` ran on Tokio worker threads, so a blocking listener starved heartbeats, the telemetry stream, the receives of every queue and the application. Java runs it on a dedicated executor of `consumptionThreadCount` threads. - New `ConsumptionExecutor`: runs the listener through `tokio::task::spawn_blocking`, at most `consumption_thread_count` messages at a time per consumer (a `Semaphore`), so the consumer uses a bounded number of the shared blocking threads whatever the number of queues and workers. - New `PushConsumerOption::consumption_thread_count` / `set_consumption_thread_count`, default **20** like Java `PushConsumerBuilderImpl`. `PushConsumer::new` (and `new_with_client`, used by `LitePushConsumer`) return `ErrorKind::Config` for 0. - Panic handling is unchanged: `consume_message` still catches a listener panic and treats it as `FAILURE`. #### Notes for reviewers - `MessageListener` (public type) is unchanged. `PushConsumer` now keeps an `Arc<ConsumptionExecutor>` internally; the test helper `new_executor` wraps listeners in tests. - Behavior change: each worker still consumes one message at a time, but at most `consumption_thread_count` listener calls run at the same time across all queues and workers of the consumer. Before, the bound was implicitly the number of runtime threads; with many queues the new default (20) can be lower than the number of workers (4 per queue), which is the point of the option. `Handle::current()` still works inside the listener: `spawn_blocking` threads carry the runtime context. - Conflicts with #1446's PR in the worker's consume call; see #1445. ### How Did You Test This Change? **Unit tests**, fail on `master`: - `test_blocking_listener_does_not_starve_the_runtime`: 1 runtime thread, listener blocks 300 ms; another task must run within 100 ms (master: 299 ms). - `test_listener_concurrency_is_bounded_by_consumption_thread_count`: 8 tasks share an executor of 3; the peak number of concurrent listener calls must be exactly 3 (checked to fail with an unbounded semaphore: 8). - `test_push_consumer_rejects_zero_consumption_thread_count`, `conf_push_consumer_option_consumption_thread_count` (default 20). **End-to-end** (mock proxy over real gRPC, 3 runs): | | `master` | this PR | |---|---|---| | 2 runtime threads, listener blocks 10 ms: unrelated 10 ms timer lateness p99 | 81.9 ms | **1.9 ms** | | same: consumed msg/s | 188 | **1,176** | | 16 queues × 4 workers, listener blocks 10 ms: peak concurrent listener calls | 4 (= runtime threads) | **20** | ``` $ cargo test # lib: 154 passed, 0 failed (master: 150) $ cargo clippy --all-features -- -D warnings # clean $ cargo fmt --check # clean $ RUSTDOCFLAGS='--cfg docsrs' cargo doc --no-deps --all-features # ok $ cargo build --examples --all-features # ok $ cargo +1.74.0 check --all-features --tests --examples # ok (MSRV) ``` --- Part of #1445 (overview, end-to-end benchmark and merge order). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
