blocks|key|2937763|text|我已经这样做了，实际上并不是那么难。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|2937764|方法是对firstRequestsObservable中的每个第一个请求(偏移量为0)进行建模。为了简单起见，您可以将其作为PublishSubject，在其中调用onNext()以提供下一个请求，但还有更智能的非主题方法(例如，如果在单击按钮时完成请求，则requestObservable是通过某些操作符映射的clickObservable+)。|offset|length|style|CODE|2937765|一旦您有了firstRequestsObservable，您就可以从firstRequestsObservable通过flatMapping执行responseObservable，依此类推，以进行服务调用。|2937766|现在诀窍来了:创建另一个从responseObservable映射的名为subsequentRequestsObservable的可观察对象，增加偏移量(为此，最好在响应数据中包括原始请求的偏移量)。引入此可观察对象后，现在必须更改responseObservable的定义，以便它也依赖于subsequentRequestsObservable。然后，您将获得一个循环依赖，如下所示：|2937767|firstRequestsObservable+->+responseObservable+->+subsequentRequestsObservable+->+responseObservable+->+subsequentRequestsObservable+->+...|2937768|要打破这个循环，您可能希望在subsequentRequestsObservable的定义中包含一个filter运算符，过滤掉偏移量将超过“总”限制的情况。循环依赖还意味着您需要将其中一个作为主题，否则将无法声明可观察对象。我推荐responseObservable作为这个主题。|2937769|因此，总而言之，您首先将responseObservable初始化为一个主题，然后声明firstRequestsObservable，然后将subsequentRequestsObservable声明为通过某些运算符传递responseObservable的结果。然后，可以通过使用onNext来“输入”responseObservable。|2937770|entityMap^0|0|2A|8|0|5|N|Y|N|20|I|0|D|I|10|S|38|I|41|S|0|0|E|S|1E|6|37|I|0|3X|6|0^^$0|@$1|2|3|4|5|6|7|T|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|U|8|@$D|V|E|W|F|G]]|9|@]|A|$]]|$1|H|3|I|5|6|7|X|8|@$D|Y|E|Z|F|G]|$D|10|E|11|F|G]|$D|12|E|13|F|G]]|9|@]|A|$]]|$1|J|3|K|5|6|7|14|8|@$D|15|E|16|F|G]|$D|17|E|18|F|G]|$D|19|E|1A|F|G]|$D|1B|E|1C|F|G]]|9|@]|A|$]]|$1|L|3|M|5|6|7|1D|8|@]|9|@]|A|$]]|$1|N|3|O|5|6|7|1E|8|@$D|1F|E|1G|F|G]|$D|1H|E|1I|F|G]|$D|1J|E|1K|F|G]]|9|@]|A|$]]|$1|P|3|Q|5|6|7|1L|8|@$D|1M|E|1N|F|G]]|9|@]|A|$]]|$1|R|3|-4|5|6|7|1O|8|@]|9|@]|A|$]]]|S|$]]

I've done this and it's actually not that hard.

The approach is to model every first request (offset 0) in a firstRequestsObservable. To make it easy, you can make this as a PublishSubject where you call <code>onNext()</code> to feed in the next request, but there are smarter non-Subject ways of doing it (e.g., if requests are done when a button is clicked, then the requestObservable is the clickObservable mapped through some operators).

Once you have <code>firstRequestsObservable</code> in place, you can make <code>responseObservable</code> by flatMapping from <code>firstRequestsObservable</code> and so forth, to make the service call.

Now here comes the trick: make another observable called <code>subsequentRequestsObservable</code> which is mapped from <code>responseObservable</code>, incrementing the offset (for this purpose it's good to include, in the response data, the offset of the originating request). Once you introduce this observable, you now have to change the definition of <code>responseObservable</code> so that it depends also on <code>subsequentRequestsObservable</code>. You then get a circular dependency like this:

firstRequestsObservable -> responseObservable -> subsequentRequestsObservable -> responseObservable -> subsequentRequestsObservable -> ...

To break this cycle, you probably want to include a <code>filter</code> operator in the definition of <code>subsequentRequestsObservable</code>, filtering out those cases where the offset would pass the "total" limit. The circular dependency also means that you need to have one of those being a Subject, otherwise it would be impossible to declare the observables. I recommend <code>responseObservable</code> to be that Subject.

So, all in all, you first initialize responseObservable as a Subject, then declare firstRequestsObservable, then declare subsequentRequestsObservable as the result of passing responseObservable through some operators. responseObservable can then be "fed in" by using <code>onNext</code>.

blocks|key|2725551|text|因此，如果这是单向分页，下面是一种您可以尝试的模式。这段代码尚未运行或编译，但我已经尝试过过度注释，以便解释发生了什么。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|2725552|private+static+final+int+LIMIT+=+50;

//+Given:+Returns+a+stream+of+dummy+event+objects+telling+us+when
//+to+grab+the+next+page.+This+could+be+from+a+click+or+wherever.
Observable<NextPageEvent>+getNextPageEvents();++

//+Given:
//+The+search+query+keywords.+Each+emission+here+means+a+new+top-level
//+request;
Observable<String>+queries;

queries.switchMap((query)+->+getNextPageEvents()
++++++++//+Ignore+'next+page'+pokes+when+unable+to+take+them.
++++++++.onBackPressureDrop()
++++++++//+Seed+with+the+first+page.
++++++++.startWith(new+NextPageEvent())
++++++++//+Increment+the+page+number+on+each+request.
++++++++.scan(0,+(page,+event)+->+page+%2B+1)+
++++++++//+Request+the+page+from+the+server.
++++++++.concatMap((page)+->+getUsersByKeyword(query,+LIMIT,+LIMIT+*+page)
++++++++++++++++//+Unroll+Observable<List<User>+into+Observable<User>
++++++++++++++++.concatMap((userList)+->+Observable.from(userList))
++++++++++++++++.retryWhen(/**+Insert+your+favorite+retry+logic+here.+*/))
++++++++//+Only+process+new+page+requests+sequentially.
++++++++.scheduleOn(Schedulers.trampoline())
++++++++//+Trampoline+schedules+on+the+'current+thread',+so+we+make+sure+that's
++++++++//+a+background+IO+thread.
++++++++.scheduleOn(Schedulers.io());|code-block|syntax|javascript|2725553|这应该会让“下一页事件”信号每次都会触发下一页数据的加载，并且在加载页面时遇到错误时不会跳转页面。如果它接收到新的搜索查询，它还会在顶级完全重新启动。如果我(或其他人？)有时间，我想检查一下我对弹床和背压的假设，并确保它阻止了在加载页面时过早获取下一个页面的任何尝试。|2725554|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|L|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|M|8|@]|9|@]|A|$]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

So, if this is one-way paging, here's a pattern you might try. This code hasn't been run or compiled, but I've tried to over-annotate in order to explain what's going on.

<pre class="lang-java prettyprint-override"><code>private static final int LIMIT = 50;

// Given: Returns a stream of dummy event objects telling us when
// to grab the next page. This could be from a click or wherever.
Observable&lt;NextPageEvent&gt; getNextPageEvents(); 

// Given:
// The search query keywords. Each emission here means a new top-level
// request;
Observable&lt;String&gt; queries;

queries.switchMap((query) -&gt; getNextPageEvents()
 // Ignore 'next page' pokes when unable to take them.
 .onBackPressureDrop()
 // Seed with the first page.
 .startWith(new NextPageEvent())
 // Increment the page number on each request.
 .scan(0, (page, event) -&gt; page + 1) 
 // Request the page from the server.
 .concatMap((page) -&gt; getUsersByKeyword(query, LIMIT, LIMIT * page)
 // Unroll Observable&lt;List&lt;User&gt; into Observable&lt;User&gt;
 .concatMap((userList) -&gt; Observable.from(userList))
 .retryWhen(/** Insert your favorite retry logic here. */))
 // Only process new page requests sequentially.
 .scheduleOn(Schedulers.trampoline())
 // Trampoline schedules on the 'current thread', so we make sure that's
 // a background IO thread.
 .scheduleOn(Schedulers.io());
</code></pre>

That should let the 'next page events' signal trigger a load of the next page's data each time, as well as not jumping pages should it encounter an error loading one. It also restarts completely at the top level if it receives a new search query. If I (or somebody else?) has time, I'd like to check my assumptions about the trampoline and backpressure and make sure it blocks any attempt to prematurely fetch the next page while one is loading.

blocks|key|146532|text|为了清楚起见，我假设你的问题更多的是如何在你的安卓应用程序中应用RxJava，而不是在后端(尽管也可以应用，但不像前端那样典型)。除了干净的代码构造之外，我不太确定这个示例是否是使用反应式函数编程(RFP)模型的典型用例。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|146533|下面的每一条流都是可观测的。就我个人而言，我认为它是流的，因为它很容易对事件进行推理。分页流可以用6个流来表示：|146534|firstPageStream+-f------------------->+//+this+is+actually+to+produce+very+first+event+to+obtain+the+first+page+result.+The+event+can+come+from+a+touch+in+one+of+the+screen+that+navigates+to+this+list+screen.
nextPageStream++-----n-------n------->+//+this+is+the+source+of+events+coming+from+Next+button+'touch'+actions
prevPageStream++---------p-------p--->+//+this+is+the+source+of+events+coming+from+Previous+button+'touch'+actions
requestStream+++-r---r---r---r---r--->+//+this+is+to+consume+the+signal+from+3+streams+above+and+spawn+events+(r)+which+create+a+query+and+pagination+details+i.e.:+offset+and+limit
responseStream++-R---R---R---R---R--->+//+this+is+to+take+each+r+and+invoke+your+web+service+getUsersByKeyword()+then+spawn+the+response+(R)|code-block|syntax|javascript|146535|上面的流可以用下面的伪代码(JS风格)表示(很容易转换成RxJava或其他语言)|146536|firstPageStream+=+Observable.just(0);+//+offset=0
nextPageStream+=+Observable.fromEvent(nextButton,+'touch')
+++++++++++++++++++++++++++.map(function()+{
++++++++++++++++++++++++++++++offset+=+offset+%2B+limit;
++++++++++++++++++++++++++++++return+offset;
++++++++++++++++++++++++++++});
prevPageStream+=+Observable.fromEvent(prevButton,+'touch')
+++++++++++++++++++++++++++.map(function()+{
++++++++++++++++++++++++++++++offset+=+offset+-+limit;+//+TODO+some+check+here
++++++++++++++++++++++++++++++return+offset;
++++++++++++++++++++++++++++});
requestStream+=+Observable.merge(firstPageStream,+nextPageStream,+prevPageStream)
++++++++++++++++++++++++++++.map(function(offsetValue)+{
+++++++++++++++++++++++++++++++return+{offset+:+offsetValue,+limit:+limit};
++++++++++++++++++++++++++++});
responseStream+=+requestStream.flatMap(function(pagination)+{
++++++++++++++++++++++++++++++++return+webservice.getUsersByKeyword(query,+pagination.offset,+pagination.limit);+//+assume+this+is+async+response
++++++++++++++++++++++++++++++});
responseStream.subscribe(function(result)+{
++//+use+result+to+render+the+display
});|146537|PS1:我没有测试过上面的代码。我正在学习RFP，所以我只是试着写下来，以被动的方式思考。欢迎任何建议。|146538|PS2:在解释反应式流的方式上，我深受https://gist.github.com/staltz/868e7e9bc2a7b8c1f754的影响。|offset|length|146539|entityMap|0|LINK|mutability|MUTABLE|url|https://gist.github.com/staltz/868e7e9bc2a7b8c1f754^0|0|0|0|0|0|0|J|1F|0|0^^$0|@$1|2|3|4|5|6|7|10|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|11|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|12|8|@]|9|@]|A|$G|H]]|$1|I|3|J|5|6|7|13|8|@]|9|@]|A|$]]|$1|K|3|L|5|F|7|14|8|@]|9|@]|A|$G|H]]|$1|M|3|N|5|6|7|15|8|@]|9|@]|A|$]]|$1|O|3|P|5|6|7|16|8|@]|9|@$Q|17|R|18|1|19]]|A|$]]|$1|S|3|-4|5|6|7|1A|8|@]|9|@]|A|$]]]|T|$U|$5|V|W|X|A|$Y|Z]]]]

To be clear, I assume your questions is more how to apply RxJava in your Android application rather than on the back end (although it's possible to apply too but not as typical as the front end). 
And i'm not very sure if this example is a typical use case for using reactive functional programming (RFP) model except for the clean code constructs.

Each stream below is an Observable. Personally, I would like to think it as stream as it's easy to reason about the events.
Pagination flow can be represented by 6 streams:

<pre><code>firstPageStream -f-------------------&gt; // this is actually to produce very first event to obtain the first page result. The event can come from a touch in one of the screen that navigates to this list screen.
nextPageStream -----n-------n-------&gt; // this is the source of events coming from Next button 'touch' actions
prevPageStream ---------p-------p---&gt; // this is the source of events coming from Previous button 'touch' actions
requestStream -r---r---r---r---r---&gt; // this is to consume the signal from 3 streams above and spawn events (r) which create a query and pagination details i.e.: offset and limit
responseStream -R---R---R---R---R---&gt; // this is to take each r and invoke your web service getUsersByKeyword() then spawn the response (R)
</code></pre>

The above streams can be represented in pseudo code (JS style) below (it would be fairly easy to translate to RxJava or other languages)

<pre><code>firstPageStream = Observable.just(0); // offset=0
nextPageStream = Observable.fromEvent(nextButton, 'touch')
 .map(function() {
 offset = offset + limit;
 return offset;
 });
prevPageStream = Observable.fromEvent(prevButton, 'touch')
 .map(function() {
 offset = offset - limit; // TODO some check here
 return offset;
 });
requestStream = Observable.merge(firstPageStream, nextPageStream, prevPageStream)
 .map(function(offsetValue) {
 return {offset : offsetValue, limit: limit};
 });
responseStream = requestStream.flatMap(function(pagination) {
 return webservice.getUsersByKeyword(query, pagination.offset, pagination.limit); // assume this is async response
 });
responseStream.subscribe(function(result) {
 // use result to render the display
});
</code></pre>

PS1: I have not tested the above code. I'm learning RFP so I just try to think in reactive way by writing down. Welcome any suggestion.

PS2: I'm highly influenced by <a href="https://gist.github.com/staltz/868e7e9bc2a7b8c1f754" rel="nofollow">https://gist.github.com/staltz/868e7e9bc2a7b8c1f754</a> for the way to explain reactive streams.

blocks|key|149171|text|这是硬摇滚！)所以我们有对网络的请求：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|149172|getUsersByKeyword(String+query,+int+limit,+int+offset)|offset|length|style|CODE|149173|例如，此请求返回|149174|List<+Result+>|149175|如果我们使用RetroFit进行联网，该请求将如下所示：|149176|Observable<+List<+Result+>>+getUsersByKeyword(String+query,+int+limit,+int+offset)|149177|因此，我们希望从服务器获取所有Result。|149178|所以它看起来就像这样|149179|int+page+=+50;
int+limit+=+page;
Observable
++++++++++++++++.range(0,+Integer.MAX_VALUE+-+1)
++++++++++++++++.concatMap(new+Func1<Integer,+Observable<List<Result>>>()+{
++++++++++++++++++++@Override
++++++++++++++++++++public+Observable<List<Result>>+call(Integer+integer)+{
++++++++++++++++++++++++return+getUsersByKeyword(query,+integer+*+page,+limit);
++++++++++++++++++++}
++++++++++++++++})
++++++++++++++++.takeWhile(new+Func1<List<Result>,+Boolean>()+{
++++++++++++++++++++@Override
++++++++++++++++++++public+Boolean+call(List<Result>+results)+{
++++++++++++++++++++++++return+!results.isEmpty();
++++++++++++++++++++}
++++++++++++++++})
++++++++++++++++.scan(new+Func2<+List<Result>,+List<Result>,+List<Result>>()+{
++++++++++++++++++++@Override
++++++++++++++++++++public+List<Result>+call(List<Result>+results,+List<+Result>+results2)+{
++++++++++++++++++++++++List<Result>+list+=+new+ArrayList<>();
++++++++++++++++++++++++list.addAll(results);
++++++++++++++++++++++++list.addAll(results2);
++++++++++++++++++++++++return+list;
++++++++++++++++++++}
++++++++++++++++})
++++++++++++++++.last()
++++++++++++++++.subscribe(new+Subscriber<List<Result>>()+{
++++++++++++++++++++@Override
++++++++++++++++++++public+void+onCompleted()+{
++++++++++++++++++++}

++++++++++++++++++++@Override
++++++++++++++++++++public+void+onError(Throwable+e)+{

++++++++++++++++++++}

++++++++++++++++++++@Override
++++++++++++++++++++public+void+onNext(List<Results>+results)+{
++++++++++++++++++++}
++++++++++++++++});|code-block|syntax|javascript|149180|代码已经测试过了！|149181|entityMap^0|0|0|1I|0|0|0|E|0|0|0|2A|0|F|6|0|0|0|0^^$0|@$1|2|3|4|5|6|7|12|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|13|8|@$D|14|E|15|F|G]]|9|@]|A|$]]|$1|H|3|I|5|6|7|16|8|@]|9|@]|A|$]]|$1|J|3|K|5|6|7|17|8|@$D|18|E|19|F|G]]|9|@]|A|$]]|$1|L|3|M|5|6|7|1A|8|@]|9|@]|A|$]]|$1|N|3|O|5|6|7|1B|8|@$D|1C|E|1D|F|G]]|9|@]|A|$]]|$1|P|3|Q|5|6|7|1E|8|@$D|1F|E|1G|F|G]]|9|@]|A|$]]|$1|R|3|S|5|6|7|1H|8|@]|9|@]|A|$]]|$1|T|3|U|5|V|7|1I|8|@]|9|@]|A|$W|X]]|$1|Y|3|Z|5|6|7|1J|8|@]|9|@]|A|$]]|$1|10|3|-4|5|6|7|1K|8|@]|9|@]|A|$]]]|11|$]]

It was hard rock!)
So we have request to network: 
<code>getUsersByKeyword(String query, int limit, int offset) </code> 
and this request return for example 
<code> List&lt; Result > </code> 
If we use RetroFit for networking that request will look: 
<code>Observable&lt; List&lt; Result >> getUsersByKeyword(String query, int limit, int offset) </code> 
As result we want to get all <code>Result</code> from server. 
So it will look like this 

<pre><code>int page = 50;
int limit = page;
Observable
 .range(0, Integer.MAX_VALUE - 1)
 .concatMap(new Func1&lt;Integer, Observable&lt;List&lt;Result&gt;&gt;&gt;() {
 @Override
 public Observable&lt;List&lt;Result&gt;&gt; call(Integer integer) {
 return getUsersByKeyword(query, integer * page, limit);
 }
 })
 .takeWhile(new Func1&lt;List&lt;Result&gt;, Boolean&gt;() {
 @Override
 public Boolean call(List&lt;Result&gt; results) {
 return !results.isEmpty();
 }
 })
 .scan(new Func2&lt; List&lt;Result&gt;, List&lt;Result&gt;, List&lt;Result&gt;&gt;() {
 @Override
 public List&lt;Result&gt; call(List&lt;Result&gt; results, List&lt; Result&gt; results2) {
 List&lt;Result&gt; list = new ArrayList&lt;&gt;();
 list.addAll(results);
 list.addAll(results2);
 return list;
 }
 })
 .last()
 .subscribe(new Subscriber&lt;List&lt;Result&gt;&gt;() {
 @Override
 public void onCompleted() {
 }

 @Override
 public void onError(Throwable e) {

 }

 @Override
 public void onNext(List&lt;Results&gt; results) {
 }
 });
</code></pre>

Code was TESTED!

I'm looking into converting my android app to use Rxjava for network requests. I currently access a webservice similar to:

<pre><code>getUsersByKeyword(String query, int limit, int offset)
</code></pre>

As I understand it, Observables are a "push" rather than a "pull" interface. So here's how I understand things to work out:

<ul>
<li>app registers with service, getting Observable for query</li>
<li>results are pushed to app</li>
<li>app deals with results</li>
<li>when app wants more results ...?</li>
</ul>

This is where things break down for me. Previously I would just ask the webservice for exactly what I want, make the query again with the offset. But in this case that would involve creating another Observable and subscribing to it, kind of defeating the point.

How should I handle paging in my app? (It's an android app, but I don't think that is relevant).

How to handle paging with RxJava?

服务器

我正在考虑将我的android应用程序转换为使用Rxjava来处理网络请求。我目前访问的webservice服务类似于：getUsersByKeyword(String query, int limit, int offset)据我所知，观察体是一个“推”而不是“拉”的界面。所以我是这样理解事情的：应用程序向服务注册，...

问如何使用RxJava处理分页？
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何使用RxJava处理分页？EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何使用RxJava处理分页？
EN