نوشته های Mohammad

منیژه‌ی جدید رسید!!

Mohammad — Mon, 09 May 2022 12:22:53 +0430

سخن‌گو جدید نشان با کلامی شیواتر منتشر شد.نسخه‌ی جدید نشان با سخن‌گو هوشمند جدید منتشر شد. «منیژه‌ی جدید» نسبت به قبل لحن شیوا‌تری داره و طنین صداش پشت فرمون بهتر به گوش می‌رسه. همچنین اسم خیابان‌ها رو نسبت به نسخه‌ی قدیمی منیژه خیلی بهتر و درست‌تر بیان می‌کنه. در نسخه‌ی جدید قابلیت‌های یادگیری و بهبود الگوریتم نسبت به قبل بهتر شده. این باعث می‌شه اگر هم جایی نامی اشتباه تلفظ می‌شد راحت‌تر و در زمانی کوتاه بتونیم بیان درست رو به الگوریتم آموزش بدیم.همون‌طور که می‌دونید بیان دقیق اسم کوچه‌ها و خیابان‌ها جز ویژگی‌های نخستین نسخه‌ی نشان بود - که دی ماه ۹۶ منتشر شد. و تا مدت‌ها نشان تنها مسیریابی بود که اسم دقیق خیابان‌ها رو اعلام می‌کرد.نسخه‌ی اولیه سخن‌گو نشان «فرهاد و نسترن» بود. فرهاد و نسترن لحن خوبی داشتند اما یک مشکل داشتند اون هم این‌که نام برخی خیابان‌ها رو درست تلفظ نمی‌کردند. به همین دلیل بعد از مدتی «بیژن و منیژه» جاشون رو گرفتند. سخن‌گوهای جدید الگوریتم‌های هوشمند بهتری داشتند و نام‌های مختلف کوچه‌ها و خیابان‌ها در شهرهای مختلف کشور رو خیلی دقیق و خوب تلفظ می‌کردند. اما لحن‌شون رباتی و کمی خشک بود. همین دلیلی بود که برخی کاربرها صداهای استودیویی - و غیر هوشمند - نظیر «فرنگیس و فریدون» رو به صدای هوشمند اما رباتی بیژن و منیژه ترجیح بدن، طوری که طرفدارهای فرنگیس از طرفدارهای منیژه بیشتر شد.همین باعث شد که یکی از اهداف ما در سال گذشته بهبود کیفیت صدای سخن‌گوهای هوشمند نشان باشه. «منیژه‌‌ی جدید» نتیجه‌ی این تلاش است. منیژه‌ی جدید اسم خیابان‌ها رو خیلی بهتر از قبل و البته با لحنی ملایم‌تر و دلنشین‌تر بیان می‌کنه. راستی این رو هم اضافه کنم که «بیژن جدید» هم در مراحل نهایی کارهای آموزشی و فنی است و امیدوارم تا آخر بهار اون هم منتشر بشه. بعدش به تدریج با بیژن و منیژه قدیمی برای همیشه خداحافظی می‌کنیم.اما علاوه بر بهبود چشم‌گیر سخن‌گوهای هوشمند، نشان مجموعه‌ای از سخن‌گوهای استودیویی را برای زبان‌های مختلف سرزمین پهناورمون ارائه می‌ده. منظور از سخن‌گو استودیویی اینه که پشتش یک الگوریتم هوشمند وجود نداره. یک خانم یا یک آقای گوینده‌ی حرفه ای در یک استودیو، دستور و فرمان‌های مختلف را ضبط کردن و صداهای ضبط شده در زمان مسیریابی پخش می‌شه. وقتی هوش مصنوعی نباشه اسم خیابان‌ها هم تلفظ نمی‌شه. چهار سخنگوی فارسی نشان شامل فرنگیس و فریدون و ایران‌‌دخت و ایرج هست. آیلار و آراس سخن‌گوهای استودیویی ترکی آذری، ژینا و ژیوار سخن‌گوهای کردی و اسماء و سمیر هم سخن‌گوهای عربی نشان هستند.امیدواریم لحن و کلام «منیژه‌ی جدید» در کنار موسیقی‌های «رادیو نشان»، چه در ترافیک‌های سخت و طولانی شهرها باشید و یا جاد‌ه‌ها زیبا و در حال سفر، خاطره‌های خوبی را از مسیر براتون بسازه.

RxJava آموزش برنامه نویسی ریکتیو در جاوا - بخش دوم

Mohammad — Fri, 13 Dec 2019 12:00:51 +0330

آشنایی با توابع ObserveOn , SubscribeOn :یک Observable بعنوان منبع داده شروع به تولید و انتشار داده ها میکند. قانون کلی حاکم بر Rx این است که در هر لحظه فقط یک داده مجاز به ورود به تابع onNext در subscription می باشد. بصورت پیش فرض تولید و انتشار داده ها در Observable در thread ای که تابع subscribe در ان ایجاد شده است شروع بکار و تولید و انتشار داده ها می کند.public static void main(String[] args) {Observable source = Observable.just("Alpha","Beta","Gamma");Observable lengths = source.map(String::length);lengths.subscribe(l -> System.out.println("Received " + l +" on thread " + Thread.currentThread().getName()));}//……output………Received 5 on thread mainReceived 4 on thread mainReceived 5 on thread mainدر مثال بالا تابع subscribe در thread اصلی برنامه شروع بکار کرده و بنابراین کل عملیات در thread اصلی برنامه و بصورت blocking انجام می شود. در این مثال تا چون عملیات در thread اصلی برنامه انجام می شود تا زمانی که تمام داده ها توسط observable منتشر نشوند و observable سیگنال onCompleted را ارسال نکند برنامه خاتمه پیدا نمیکند. در صورتی که بخواهیم کل عملیات تولید و انتشار داده توسط observable در thread ای غیر از thread اصلی برنامه اتفاق بیفتد میتوانیم از تابع SubscribeOn() و مشخص نمودن یک Scheduler استفاده نماییم. با اینکار تولید و انتشار داده ها در thread غیر از thread main انجام می شود. اگر عملیاتی محاسباتی در حال انجام باشد از Schedulers.computation() استفاده مینماییم. این scheduler بصورت پیش فرض به تعداد core های cpu ، thread در اختیار برنامه قرار می دهد.public static void main(String[] args) {Observable source = Observable.just("Alpha", "Beta", "Gamma");Observable lengths = source.subscribeOn(Schedulers.computation()).map(String::length);lengths.subscribe(sum -> System.out.println("Received " + sum +" on thread " + Thread.currentThread().getName()));}// output……Received 5 on thread RxComputationThreadPool-1Received 4 on thread RxComputationThreadPool-1Received 5 on thread RxComputationThreadPool-1اگر مثال بالا را اجرا کنید قبل از نمایش نتیجه، برنامه خاتمه پیدا میکند به این دلیل که محاسبات بالا در یک thread دیگر اتفاق افتاده و thread اصلی برنامه قبل از اینکه thread محاسباتی خروجی را نمایش دهد به اتمام می رسد. یک برداشت اشتباه زمانی که از SubscribeOn استفاده می کنیم این است که برنامه باید بصورت خودکار بصورت پارالل اجرا شود یعنی در مثال بالا ارسال و نمایش اطلاعات بصورت موازی و در چند thread جداگانه انجام شود، ولی همانطور که از خروجی پیداست تمامی محاسبات و خروجی در یک thread یکسان انجام شده است و این رفتار پیش فرض Rx می باشد.مکان استفاده از SubscribeOn() تفاوتی در رفتار ان ایجاد نمیکند، در واقع حتی اگر چند بار از این تابع در کد خود استفاده کنید فقط نسخه ای که به Observable نزدیکتر است مورد استفاده قرار گرفته و مابقی SubscribeOn ها نادیده گرفته می شوند.Source1.map().subscribeOn(Scheduler.IO).filter.subscribeOn(Scheduler.Computation())در مثال بالا عملیات در IO thread انجام می شود و دومین SubscribeOn نادیده گرفته می شود. تابعی مثل Observable.interval() بصورت پیش فرض و داخلی یک تابع SubscribeOn با Schedulers.Computation دارد و این تابع بصورت پیش فرض عملیاتش را یک thread محاسباتی انجام می دهد. بطور کلی تابع SubscribeOn به Observable اعلام میکند که در چه thread ای باید عملیات خود را انجام دهد.ObserveOn() تابع بعضی مواقع پیش می آید که در زنجیره محاسبات و عملگرهای که روی یک Observableایجاد کرده اید ادامه کار را روی thread دیگری در scheduler دیگری اجرا کنید. بعنوان مثال تولید داده در IO Scheduler در حال انجام است و در عملگرهای بعدی که روی داده های منتشر شده قرار است انجام شود محاسبات را بخواهیم روی یک Thread از Computation Scheduler انجام دهیم. در این حالت میتوانیم از تابع ObserveOn استفاده نماییم. در مثال زیر چونکه از تابع SubscribeOn استفاده نکرده ایم انتشار داده ها در Thread اصلی برنامه انجام می شود و قبل از عملگر map از تابع ObserveOn استفاده کرده ایم و عملا ادامه زنجیره تا انتها روی Thread دیگری از Computation Schedulerانجام می شود. در واقع به هر تعدادی که بخواهیم میتوانیم از تابع ObserveOn در زنجیره توابع Rx استفاده کنیم و در هر بار استفاده از محل استفاده به بعد Thread اجرای زنجیره تغییر میکند و توابع از قبل از ObserveOn تحت تاثیر تغییر Thread قرار نمیگیرند.public static void main(String[] args) {Observable source = Observable.range(1,10);source.map(i -> i * 100).doOnNext(i -> System.out.println("Emitting " + i+ " on thread " + Thread.currentThread().getName())).observeOn(Schedulers.computation()).map(i -> i * 10).subscribe(i -> System.out.println("Received " + i + " on thread "+ Thread.currentThread().getName()));sleep(3000);}// output…..Emitting 100 on thread mainEmitting 200 on thread mainEmitting 300 on thread mainEmitting 400 on thread mainEmitting 500 on thread mainEmitting 600 on thread mainEmitting 700 on thread mainEmitting 800 on thread mainEmitting 900 on thread mainEmitting 1000 on thread mainReceived 1000 on thread RxComputationThreadPool-3Received 2000 on thread RxComputationThreadPool-3Received 3000 on thread RxComputationThreadPool-3Received 4000 on thread RxComputationThreadPool-3Received 5000 on thread RxComputationThreadPool-3Received 6000 on thread RxComputationThreadPool-3Received 7000 on thread RxComputationThreadPool-3Received 8000 on thread RxComputationThreadPool-3Received 9000 on thread RxComputationThreadPool-3Received 10000 on thread RxComputationThreadPool-3انتخاب Scheduler : اسکجولر های متنوعی در rx برای انجام کارهای متفاوت تعبیه شده است. Schedulers.io() که برای انجام کارهای مرتبط با IO بهینه شده است و بصورت پیش فرض thread های داخل این scheduler کش شده و مورد استفاده مجدد قرار میگرند. Schedulers.newThread() که به ازای هر subscription یک thread جدید ایجاد میکند. در هنگام استفاده از این دو scheduler باید با احتیاط زیادی عمل کنید چون میتوانند منجر به تولید تعداد زیادی thread شده و کارایی برنامه را کاهش خواهند داد. برای کارهای محاسباتی میتوان از Schedulers.computation() استفاده نمایید که حداکثر تعداد thread های آن به تعداد core های cpu محدود شده است. همچنین از Schedulers.from() نیز میتوان استفاده کرد و در واقع Executor مد نظر خود با تعداد thread های مد نظر خود را ایجاد نمایید.int computationalThreadCount = Runtime.getRuntime().availableProcessors() + 1; ExecutorService executor = Executors.newFixedThreadPool(computationalThreadCount); Scheduler schedulerComputation = Schedulers.from(executor);

RxJava آموزش برنامه نویسی ریکتیو در جاوا - بخش اول

Mohammad — Tue, 05 Nov 2019 08:15:01 +0330

سلام به همه دوستان. توی اولین پستم میخام مطالعاتی که در مورد RxJava از منابع مختلف داشتم رو براتون به اشتراک بزارم. امیدوارم براتون مفید باشه.آرایکس جاوا (RxJava : Reactive Extension for JVM) یک کتابخانه برای برنامه نویسی غیرهمزمان (asynchronous) و مبتنی بر رخداد (event-based) با استفاده از الگوی Observer می باشد. این کتابخانه مزیتهای بسیاری برای ما ایجاد میکند که از جمله مهمترین آنها استفاده از thread ها و thread pool ها را برای برنامه نویسی راحت تر کرده و دیگر برنامه نویس نیازی نیست نگران جزییات اجرای کارهای خود بروی Thread های مجزا از thread اصلی برنامه باشد و همچنین امکان هر دو مدل برنامه نویسی همزمان و غیرهمزمان را فراهم میکند همچنین این کتابخانه، توابع بسیار زیادی برای کار بر روی داده های تولید شده توسط تولید کننده یا همان منبع داده (observable) فراهم میکند. الگوی برنامه نویسی reactive این مفهوم را بیان میکند که در زمان نیاز بتوانیم به تغییرات پاسخ دهیم به این معنی که یک منبع داده شروع به تولید داده ها میکند که به اصطلاح به آن مشاهده شونده یا تولید کننده Observable میگوییم. هر منبع تولید داده ای نیاز به یک یا چند مشاهده کننده یا مصرف کننده Observer ویا به اصطلاح Subscriber دارد و ارتباط بین مشاهده شونده و مشاهده کننده با مفهومی بنام مشترک شدن یا Subscribe برقرار می شود. به محض اینکه ارتباط بین تولید کننده و مصرف کننده توسط subscribe برقرار شود کانال ارسال داده های تولید کننده به سمت مصرف کننده برقرار شده و تولید کننده شروع به تولید داده و مصرف کننده شروع به مصرف ان می کند.برای استفاده از این کتابخانه در spring boot وابستگی زیر را به فایل pom.xml اضافه میکنیم.io.reactivex.rxjava2rxjava2.2.8شکل کلی دستورات در RxJava بصورت زیر هستش.source.operator1().operator2().operator3().subscribe(consumer);یک منبع داده، صفر یا چند عملگر میانی و در انتها مصرف کنندهسه مفهوم کلی در برنامه نویسی با RxJava :1- منبع داده (تولید کننده) Source of data2- مصرف کننده داده Consumer of data3- اتصال مصرف کننده به منبع داده Connecting Consumer to Source// defining the sourceObservable source = Observable.range(1, 5);// defining the consumerSubscriber consumer = new Subscriber() {@Overridepublic void onNext(Integer number) { System.out.println(number); }@Overridepublic void (Throwable e) { System.out.println("error"); }@Overridepublic void onCompleted() { System.out.println("completed"); }};// connecting the consumer to sourcesource.subscribe(consumer);تابع onNext می تواند اصلا فراخوانی نشود و یا یک و یا چندین بار و یا حتی بصورت بی نهایت فراخوانی شود. توابع و onCompleted() را رخدادهای پایانی یا به اصطلاح terminal events می نامند به این معنی که فقط و فقط یکبار میتوانند فراخوانی شوند.زمانی که یک terminal event فراخوانی شود استریم تولید داده observable به اتمام می رسد و هیج داده ای دیگری ارسال نخواهد شد. رخدادهای پایانی اگر استریم تولید داده بی نهایت باشد ممکن است اصلا فراخوانی نشوند.منبع داده:منبع تولید داده Observable می باشد که اعداد بین 1 تا 5 را بعنوان داده تولید خواهد کرد.مصرف کننده:مصرف کننده Subscriber داده های تولید شده می باشد. RxJava داده های تولید شده توسط تولید کننده را توابع onNext مصرف کننده ارسال میکند. در صورتی که در فرایند ارسال داده یا به اصطلاح emit کردن داده به سمت مصرف کننده هرگونه خطایی رخ دهد تابع مصرف کننده فراخوانی شده و کانال ارسال داده بسته خواهد شد. وقتی تمامی داده های تولید کننده به سمت مصرف کننده با موفقیت ارسال شد، RxJava سیگنال اتمام تولید داده را به تابع onCompleted مصرف کننده ارسال میکند.اتصال مصرف کننده به منبع داده:در مثال بالا تمام محاسبات و کدهایی که بعنوان بخشی از تولید کننده و مصرف کننده نوشته شده فقط زمانی که اتصال بین منبع داده و مصرف کننده برقرار شده باشد، اجرا خواهد شد. در واقع در مثال بالا تا زمانی که مصرف کننده ارتباطش با تولید کننده از طریق تابع subscribe برقرار نشده باشد، منبع داده هیچ داده ای را تولید نخواهد کرد و فرایند مربوطه شروع نخواهد شد.انواع منبع داده در RxJava:io.reactivex.rxjava3.core.Flowable: انتشار صفر یا بی نهایت داده با قابلیت بافر کردن داده های ارسالیio.reactivex.rxjava3.core.Observable: انتشار صفر یا بی نهایت داده بدون قابلیت بافرینگ داده های ارسالیio.reactivex.rxjava3.core.Single: منبع داده ای با قابلیت انتشار دقیقا یک داده و یا ارسال سیگنال خطاio.reactivex.rxjava3.core.Completable: منبع داده ای بدون انتشار داده و فقط ارسال سیگنال اتمام کار و یا سیگنال خطاio.reactivex.rxjava3.core.Maybe: منبع داده ای بدون انتشار داده یا دقیقا انتشار فقط یک داده و یا ارسال سیگنال خطامنبع داده Observable :این منبع داده را میتوان در واقع قلب RxJava نامید. منبع تولید داده یا رخداد (event) در RxJavaمی باشد. RxJava توابع زیادی را برای ایجاد یک منبع داده (observable) در اختیار ما قرار می دهد. انواع توابعی که برای ایجاد یک منبع داده وجود دارد را در ادامه با هم بررسی میکنیم. توابع زیر اکثراً در observable و flowable مشترک می باشند.منبع داده Flowable :این منبع داده علاوه برداشتن قابلیتهای observable بعنوان یک منبع داده قابلیتی بنام BackPressure دارد. زمانی که سرعت تولید و انتشار داده توسط یک observable بیشتر از توان مصرف کننده در مصرف داده های ارسالی باشد چه اتفاقی رخ میدهد؟ observable یک بافر با اندازه بی نهایت دارد به این معنی که تمامی داده هایی که قرار است به سمت مصرف کننده ارسال کند را بافر کرده و سپس ارسال می کند و در حالتی که مصرف کننده توان مصرف نرخ داده ارسالی را نداشته باشد خطای outOfMemoryException رخ می دهد. با اعمال قابلیت backPressure بر روی یک استریم داده میتوان داده های ارسالی را به روش دلخواه مدیریت کرد. در نسخه 1 rxJava اعمال این قابلیت بر عهده observable بود ولی از نسخه 2 rxJava این قابلیت بر عهده کلاس flowable قرار داده شده است. همچنین میتوانیم یک observable را با تابع زیر به یک flowable با قابلیت backpressure تبدیل کنیم.Observable.range(1, 1_000_000).toFlowable(BackpressureStrategy.Drop)انواع حالتهایی که backpressure در اختیار ما قرار میدهد بصورت زیر می باشد:Drop: اگر تعداد ایتمهای تولید شده از سایز بافر بزرگتر شد دور ریخته می شود.Buffer: تمامی داده های تولید شده را بافر میکند.Latest: فقط آخرین مورد را نگه می داردError: هنگامی که سرعت انتشار داده از توان مصرف بیشتر باشد خطای MissingBackpressureException ایجاد میکند.Missing: No strategy, it would throw a MissingBackpressureException sooner or later somewhere on the downstream منبع داده Single :زمانی که بخواهیم منبع داده ما دقیقا یک آیتم را به مصرف کننده ارسال کند و یا در صورت بروز خطا سیگنال error را ارسال کند از این نوع منبع داده استفاده می نماییم. این منبع داده سیگنال onComplete ارسال نمیکند.Single.fromCallable(this::returnSomething) .subscribe( (item) -> { System.out.println(item); }, (err) -> { System.out.println(err); });private int returnSomething() { return 100; }منبع داده Completable :یک منبع داده بدون تولید و انتشار هیچگونه داده ای که فقط سیگنال completionو یا errorرا صادر میکند. کاربرد آن زمانی است که یک تابعی قرار است یک کاری را انجام دهد ولی خروجی که برمیگرداند برای ما اهمیت ندارد و یا اینکه اصلا خروجی برنمیگرداند و صرفاً انجام خود کار برایمان مهم است مثلا یکسری مقداردهی های اولیه در برنامه یا مثلا بروز رسانی یک مقداری در کش و یا انجام یک سری عملیات در دیتابیس.Completable.fromAction(this::call) .subscribe( () -> { System.out.println("Complete"); }, (err) -> { System.out.println(err); });private void call() { System.out.println(1); }منبع داده Maybe :مشابه با Single با این تفاوت که امکان انتشار و تولید هیچگونه داده ای را نیز فراهم میکند. یعنی عملا ترکیبی از Completable و single می باشد.//Maybe is similar to Single only difference being that it allows for no emissions as well. Disposable disposable = Maybe.just("Hello World Maybe") .delay(2, TimeUnit.SECONDS, Schedulers.io()) .subscribeWith(new DisposableMaybeObserver() { @Override public void (Throwable e) { e.printStackTrace(); } @Override public void onSuccess(String value) { System.out.println(value); } @Override public void onComplete() { System.out.println("Done Maybe!"); } }); Thread.sleep(3000); //start observing disposable.dispose(); //no emission Maybe emptySource = Maybe.empty(); emptySource.subscribe( s -> System.out.println("Item received: from emptySource" + s), // اجرا نمی شود Throwable::printStackTrace, () -> System.out.println("Done from EmptySource") );انواع توابعی که منابع داده بالا برای تولید و انتشار داده در اختیار ما قرار می دهند:تابع just : این تابع در منابع داده زیر موجود است:observable,flowable,maybe,singleبرای ایجاد منبع داده از روی یک سری از اشیا از پیش ساخته شده و اماده میتوانیم از این تابع استفاده کنیم. این منبع داده به محض اینکه یک مصرف کننده به ان متصل شود شروع به ارسال داده ها به سمت مصرف کننده خواهد کرد. این تابع ده overload مختلف دارد که از بین 1 تا 10 ایتم را بعنوان ورودی میپذیرد.Observable observable = Observable.just("100", "mohammad", 200, 2.4); observable.subscribe( item -> System.out.print(item), error -> error.printStackTrace(), () -> System.out.println("finish") );تابع FromIterable:این تابع در منابع داده زیر موجود است:flowable,observableبرای ایجاد یک منبع تولید داده از روی یک لیست استفاده می شود. در واقع هر کدام از ایتمهای داخل لیست بصورت مجزا به سمت مصرف کننده ارسال می شوند.List list = new ArrayList<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8)); Observable observable = Observable.fromIterable(list); observable.subscribe( item -> System.out.println(item), error -> error.printStackTrace(), () -> System.out.println("Done"));تابع fromArray :این تابع در منابع داده زیر موجود است:flowable,observableبعنوان یک منبع داده از روی یک آرایه عمل کرده و هر یک از عناصر آرایه را بعنوان یک آیتم مجزا به سمت مصرف کننده ارسال میکند.از آرایه های primitive پشتیبانی نمیکند و فقط مجاز به استفاده از Wrapper type ها می باشید.Integer[] array = new Integer[10]; for (int i = 0; i < array.length; i++) { array[i] = i; } Observable observableArrays = Observable.fromArray(array); observableArrays.subscribe( item -> System.out.println(item), error -> error.printStackTrace(), () -> System.out.println("Done"));تابع fromCallable : در منابع داده زیر موجود است:flowable,observable,maybe,single,completableکارکرد این تابع زمانی است که یک تابع برای انجام کاری نوشته اید و این تابع پس از اتمام محاسبه یک نتیجه را برمیگرداند. بعنوان مثال تابعی که یک سرویس http را برای دریافت اطلاعاتی از سمت سرور فراخوانی کرده و بعد از فراخوانی آن سرویس، نتیجه را به قسمتی که تابع را فراخوانی کرده برگشت می دهد. مواردی از این قبیل با استفاده از اینترفیس callable قابل انجام می باشد.Callable callable = () -> { System.out.println("inside callable!"); return "Hello World!" // http method call or whatever }; Observable observable = Observable.fromCallable(callable); observable.subscribe( item -> System.out.println(item), error -> error.printStackTrace(), () -> System.out.println("Done") );//…. With lambda syntaxObservable.fromCallable((Callable) () -> null).subscribe(item -> System.out.println(item), error -> error.printStackTrace(), () -> System.out.println("Done"));تابع Range : در منابع داده زیر موجود است:flowable,observableبا استفاده از این تابع یک دنباله افزایشی از اعداد صحیح را میتوانید بصورت یک به یک به سمت مصرف کننده ها ارسال نمایید. تابع rangeLong برای ایجاد یک دنباله از اعداد long استفاده می شود و اینکار پیش فرض بصورت blocking و روی thread اصلی برنامه اجرا می شود.Observable.range(1,5).subscribe(item-> System.out.println(item));Observable.rangeLong(1,5).subscribe(item-> System.out.println(item));تابع Timer: بعد از گذشت مدت زمان مورد نظر یک عدد 0 را به سمت مصرف کننده ارسال کرده و سپس سیگنال complete را ارسال میکند. در مثال زیر بعد از گذشت 5 ثانیه تابع onNext مصرف کننده را با مقدار 0 فراخوانی کرده و سپس خاتمه پیدا میکند. این کار را بصورت nonBlocking و در یک thread دیگر انجام می دهد.Observable timer = Observable.timer(5, TimeUnit.SECONDS); timer.subscribe( (v) -> System.out.println(Thread.currentThread().getName() + v + " onNext!"), // RxComputationThreadPool-10 onNext! (err) -> System.out.println(err), () -> System.out.println("finish"));مثال زیر بصورت blocking و در thread اصلی برنامه با استفاده از تابع blockingSubscribe اینکار را انجام می دهد.Observable timer = Observable.timer(5, TimeUnit.SECONDS); timer.blockingSubscribe( (v) -> System.out.println(Thread.currentThread().getName() + v + " onNext!"), // main0 onNext! (err) -> System.out.println(err), () -> System.out.println("finish"));انشالا توی بخش دوم میخام در مورد توابع subscribeOn , observerOn و همچنین اجرای کدها بصورت موازی و در واقع MultiThread توی RxJava صحبت کنم.