Reactive Programming Practice in Xianyu: RxJava Refactoring and Performance Evaluation

Reactive programming is defined as programming with asynchronous data streams. It appears in frameworks such as Spring 5 and Java 9 and is considered a future direction for software development.

The model is inherently non‑blocking: actions are triggered automatically when data becomes available, reducing the need for thread‑blocking and improving system throughput.

Xianyu’s complex business logic originally relied on blocking code, causing many threads to wait for I/O (logging, RPC, HTTP, etc.). To address this, RxJava 2.0 was introduced, and key features such as the fish‑pond homepage, message list, and detail page were migrated to an asynchronous model.

Traditional Java concurrency (Executor, Future) suffers from fragmented thread pools, excessive context switches, and blocking waits. RxJava unifies thread‑pool management, dramatically increases thread utilization, and enables an almost ideal thread model where only as many threads as CPU cores are needed for computation.

The migration identified four major blocking points: application logging, HTTP requests, RPC calls, and cache read/write. As an example, log4j2 provides a lock‑free, Disruptor‑based asynchronous logger that can increase throughput tenfold compared to log4j or logback.

Three execution paradigms were demonstrated:

Serial request – a flow that first fetches an order and then its related item.

//查订单
Flowable<Order> orderFlow = Flowable.fromCallable(() -> queryOrder(orderId));
//查商品
Flowable<Item> itemFlow = orderFlow.flatMap(order -> Flowable.fromCallable(() -> queryItem(order.getId())));

Concurrent request – fetching page‑view and comment data in parallel and combining them.

Flowable<Detail> detailFlow = itemFlow.flatMap(item -> {
    //浏览量
    Flowable<Long> pvFlow = Flowable.fromCallable(() -> queryItemPv(item.getId()));
    //留言
    Flowable<Comment> commentFlow = Flowable.fromCallable(() -> queryItemComment(item.getId()));
    //合并结果
    return Flowable.zip(pvFlow, commentFlow, (pv, comment) -> buildItemDetail(item, pv, comment));
});

Cache update – side‑effect after the main flow completes. detailFlow.doOnNext(detail -> cache(detail)); Performance tests compared the original blocking implementation with the fully asynchronous reactive version. Results showed a ~50% reduction in response time (rt), stable thread counts under high QPS, and higher CPU utilization (up to 97% vs. 75% for blocking). Throughput increased by roughly 30%.

The study concludes that reactive programming with RxJava significantly improves Xianyu’s service performance, while future work will focus on eliminating remaining blocking points, leveraging a CPU‑core‑sized thread pool for all computation, and unifying the I/O thread‑pool model.