tag:blogger.com,1999:blog-6753769565491687768.post3817705035317284328..comments2019-02-21T06:29:44.506+01:00Comments on Tomasz Nurkiewicz around Java and concurrency: Small scale stream processing kata. Part 2: RxJava 1.x/2.xTomasz Nurkiewicznoreply@blogger.comBlogger3125tag:blogger.com,1999:blog-6753769565491687768.post-36612418182159439852016-11-01T16:21:05.594+01:002016-11-01T16:21:05.594+01:00I&#39;m actually considering using the same patter...I&#39;m actually considering using the same pattern as you with a custom operator hiding the budget counter state.<br /><br />Long story short, we&#39;ve actually been using Hystrix extensively but for a couple of reasons would like to retry non-blocking IO clients inside the Hystrix execution instead of as an operator on the Hystrix observable.<br /><br />As our client uses RX natively it would be very clean to deal with &quot;plain rx&quot;, but if everything else fails I&#39;ll probably look into simply using circuit breakers to acheive similar results.Billy Sjöberghttps://www.blogger.com/profile/16665575968269641445noreply@blogger.comtag:blogger.com,1999:blog-6753769565491687768.post-57827306468971457002016-11-01T15:53:10.263+01:002016-11-01T15:53:10.263+01:00Interesting problem. First of all try Hystrix that...Interesting problem. First of all try Hystrix that implements limited concurrency via thread pools or semaphores. In plain RxJava consider flatMap() taking int (to limit concurrency) or small Scheduler built from ExecutorService with limited size.Tomasz Nurkiewiczhttps://www.blogger.com/profile/05938011050162061962noreply@blogger.comtag:blogger.com,1999:blog-6753769565491687768.post-91544075542245243322016-11-01T11:17:34.092+01:002016-11-01T11:17:34.092+01:00Awesome post as usual Tomasz.
I have a use-case pr...Awesome post as usual Tomasz.<br />I have a use-case pretty similar to your cache that would be really interesting to follow up with.<br />I&#39;m trying to implement a retry budget for an async IO computation, meaning that if calls fail, only X nr of retries per second are allowed globally in order to not put more stress on the backend.<br /><br />My naive solution would be to have a global AtomicInteger that gets reset each second by an interval observable. In the IO observable I would then check and decrease that counter during the retryWhen, meaning that only the first X errors during each second would cause a retry.<br /><br />Even before I read your article my gut feeling told me that having global state referenced by two observables is a bit no-no, and I guess it should be implemented using some zip with a windowed observable + take.<br />Do you have any pointers on how to proceed?<br /><br />Love the book so far btw, awesome stuff.Billy Sjöberghttps://www.blogger.com/profile/16665575968269641445noreply@blogger.com