Reactive programming aur Project Reactor — streaming ke liye

Virtual threads ne reactive programming ko bahut si jagah optional bana diya hai. Pehle jo cheez majboori thi, ab choice hai. Lekin AI kaam mein ek jagah aisi hai jahan Reactor aaj bhi bilkul sahi tool hai: streaming.

Sabse pehle: reactive ka matlab kya hai#

Normal (blocking) code mein aap kehte hain "mujhe jawab do", aur thread wahin khada rehta hai jab tak jawab na aa jaaye.

Reactive code mein aap kehte hain "jab jawab aaye to ye kar dena", aur thread aage badh jaata hai. Aap data nahi maangte — aap us data ke aane par kya karna hai wo pehle se likh dete hain. Isi ko declarative pipeline kehte hain.

Do type hain, aur bas do:

Mono<String> answer  = ...;   // zero ya ek item, phir complete
Flux<String> tokens  = ...;   // zero se kai items, time ke saath

Ek aur zaroori baat: jab tak koi subscribe na kare, kuch chalta hi nahi. Aapne pipeline likh di lekin subscribe nahi kiya, to ek bhi line execute nahi hogi. Spring controller se Mono ya Flux return kar dijiye — framework khud subscribe kar leta hai.

Wo ek jagah jahan Reactor sach mein jeetta hai#

ChatGPT jaisa experience — jawab ek saath nahi aata, shabd-shabd karke aata hai — user ko bahut fast lagta hai. Poora paragraph 12 second baad ek saath dena aur pehla shabd 400 millisecond mein dena, technically same latency ka kaam hai, lekin mehsoos bilkul alag hota hai.

Spring AI mein streaming ek line ka farak hai:

@RestController
public class ChatController {
 
    private final ChatClient chatClient;
 
    public ChatController(ChatClient.Builder builder) {
        this.chatClient = builder.build();
    }
 
    @GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> stream(@RequestParam String question) {
        return chatClient.prompt()
                .user(question)
                .stream()
                .content();
    }
}

.call() ki jagah .stream(), aur return type String ki jagah Flux<String>. Bas. Spring is Flux ko Server-Sent Events mein badal deta hai, aur browser side par:

const source = new EventSource('/chat/stream?question=' + encodeURIComponent(q));
source.onmessage = (event) => { output.textContent += event.data; };
source.onerror = () => source.close();

Roz kaam aane wale operators#

Zyadatar AI code mein inhi paanch se kaam chal jaata hai:

Flux.fromIterable(documents)                       // list se stream banaya
    .filter(doc -> doc.text() != null)             // khaali hataye
    .map(doc -> splitter.split(doc))               // sync transform
    .flatMap(chunks -> embedAsync(chunks), 4)      // async transform, 4 ek saath
    .doOnNext(result -> log.debug("embedded {}", result.id()))
    .collectList();                                // wapas ek Mono<List<...>>

map aur flatMap ka farak yaad rakhiye, kyunki galti yahi hoti hai:

• map tab jab aapka function seedha value deta ho.
• flatMap tab jab aapka function khud Mono ya Flux deta ho.

flatMap ka doosra argument concurrency hai — upar 4 ka matlab hai ek saath chaar embedding calls se zyada nahi. Model provider ke rate limit ko izzat dene ka ye sabse aasaan tareeka hai.

Back-pressure — asli cheez yahi hai#

Back-pressure Reactor ka wo hissa hai jo normal async code mein hai hi nahi. Consumer producer ko bata sakta hai ki "mujhe abhi sirf itne items chahiye".

Ingestion pipeline mein iski value turant dikhti hai. Aap 50,000 documents padh rahe hain aur har ek ko embed kar rahe hain. Embedding API dheemi hai. Bina back-pressure ke aapki memory un documents se bhar jaayegi jo queue mein embed hone ka intezaar kar rahe hain, aur ek din OutOfMemoryError milega.

Flux.fromIterable(hugeDocumentList)
    .limitRate(64)                                   // ek baar mein 64 se zyada mat kheencho
    .flatMap(doc -> embeddingClient.embed(doc), 8)   // 8 se zyada ek saath mat bhejo
    .buffer(100)                                     // 100-100 ke batch mein store karo
    .flatMap(vectorStore::saveAll)
    .subscribe();

Ye teen lines poore pipeline ko memory-safe bana deti hain, aur aapko koi manual queue ya semaphore nahi likhna pada.

Kab reactive nahi lena chahiye#

Ye hissa sabse important hai, aur aksar chhoda jaata hai.

Simple normal (blocking) code likhiye jab:

• Aapko sirf ek jawab chahiye, stream nahi. Virtual threads is kaam ko aaram se sambhal lete hain.
• Aapka data JDBC se aa raha hai aur R2DBC par jaane ka koi plan nahi hai.
• Team ne Reactor mein debug nahi kiya hai. Reactive stack trace 60 line lambi hoti hai aur usme aapke code ki ek bhi line nahi hoti — ye seekhne mein waqt lagta hai.

Reactor lijiye jab:

• Aap tokens stream kar rahe hain (yahi sabse aam wajah hai).
• Aapko back-pressure chahiye kyunki producer consumer se tez hai.
• Aap pehle se WebFlux par hain.

Ek badhiya beech ka raasta bhi hai: poori application normal Spring MVC par rakhiye aur sirf streaming endpoint se Flux return kar dijiye. Spring MVC Flux return type ko theek se handle kar leta hai — poori application ko WebFlux par le jaane ki koi zaroorat nahi.

Aage kya#

Foundation yahin poora hota hai — Java 21, functional style, build setup, deployment, architecture aur streaming. Ab asli AI kaam shuru hota hai: Spring AI ke saath pehla ChatClient call.