rx-java
शेड्यूलर

मूल उदाहरण

अनुसूचक प्रसंस्करण इकाई के बारे में एक RxJava अमूर्त हैं। एक शेड्यूलर को एक एक्सेक्यूटर सेवा द्वारा समर्थित किया जा सकता है, लेकिन आप अपने स्वयं के शेड्यूलर कार्यान्वयन को लागू कर सकते हैं।

एक Scheduler को इस आवश्यकता को पूरा करना चाहिए:

• क्रमिक रूप से कार्य को अनौपचारिक रूप से संसाधित करना चाहिए (FIFO आदेश)
• टास्क में देरी हो सकती है

एक Scheduler : कुछ ऑपरेटरों (उदाहरण में पैरामीटर के रूप में इस्तेमाल किया जा सकता delay , या के साथ प्रयुक्त) subscribeOn / observeOn विधि।

कुछ ऑपरेटर के साथ, Scheduler का उपयोग विशिष्ट ऑपरेटर के कार्य को संसाधित करने के लिए किया जाएगा। उदाहरण के लिए, delay एक विलंबित कार्य को शेड्यूल करेगा जो अगले मान का उत्सर्जन करेगा। यह एक Scheduler जो बाद में इसे बनाए रखेगा और निष्पादित करेगा।

subscribeOn का उपयोग एक बार Observable अनुसार किया जा सकता है। यह परिभाषित करेगा कि किस Scheduler में सदस्यता का कोड निष्पादित किया जाएगा।

observeOn प्रति कई बार इस्तेमाल किया जा सकता है Observable । यह परिभाषित करेगा कि observeOn विधि के बाद परिभाषित सभी कार्यों को निष्पादित करने के लिए Scheduler का उपयोग किया जाएगा। observeOn प्रदर्शन करने में मदद करेंगे।

सदस्यता लें विशिष्ट अनुसूचक

// this lambda will be executed in the `Schedulers.io()`
Observable.fromCallable(() -> Thread.currentThread().getName())
          .subscribeOn(Schedulers.io())
          .subscribe(System.out::println); 

विशिष्ट शेड्यूलर के साथ अवलोकन करें

Observable.fromCallable(() -> "Thread -> " + Thread.currentThread().getName())
         // next tasks will be executed in the io scheduler
         .observeOn(Schedulers.io())
         .map(str -> str + " -> " + Thread.currentThread().getName())
          // next tasks will be executed in the computation scheduler
         .observeOn(Schedulers.computation())
         .map(str -> str + " -> " + Thread.currentThread().getName())
         // next tasks will be executed in the io scheduler
         .observeOn(Schedulers.newThread())
         .subscribe(str -> System.out.println(str + " -> " + Thread.currentThread().getName()));   

ऑपरेटर के साथ एक विशिष्ट समयबद्धक निर्दिष्ट करना

कुछ ऑपरेटर पैरामीटर के रूप में एक Scheduler ले सकते हैं।

Observable.just(1)
          // the onNext method of the delay operator will be executed in a new thread
          .delay(1, TimeUnit.SECONDS, Schedulers.newThread())
          .subscribe(System.out::println);

सब्सक्राइबर को प्रकाशित करें:

TestScheduler testScheduler = Schedulers.test();
EventBus sut = new DefaultEventBus(testScheduler);
TestSubscriber<Event> subscriber = new TestSubscriber<Event>();
sut.get().subscribe(subscriber);
sut.publish(event);
testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);

धागा पूल:

this.poolName = schedulerFig.getIoSchedulerName();
final int poolSize = schedulerFig.getMaxIoThreads();
final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(poolSize);
final MaxSizeThreadPool threadPool = new MaxSizeThreadPool( queue, poolSize );
this.scheduler = Schedulers.from(threadPool);

वेब सॉकेट अवलोकनीय:

final Subscription subscribe = socket.webSocketObservable()
        .subscribeOn(Schedulers.io())
        .doOnNext(new Action1<RxEvent>() {
            @Override
            public void call(RxEvent rxEvent) {
                System.out.println("Event: " + rxEvent);
            }
        })
        .subscribe();