我們來(lái)談?wù)刯ava操作cassandra分頁(yè)，需要注意的是這個(gè)分頁(yè)與我們平時(shí)所做的頁(yè)面分頁(yè)是不同的，具體有啥不同，大家耐著性子往下看。

如果一個(gè)查詢(xún)得到的記錄數(shù)太大，一次性返回回來(lái)，那么效率非常低，并且很有可能造成內(nèi)存溢出，使得整個(gè)應(yīng)用都奔潰。所以了，驅(qū)動(dòng)對(duì)結(jié)果集進(jìn)行了分頁(yè)，并返回適當(dāng)?shù)哪骋豁?yè)的數(shù)據(jù)。

一、設(shè)置抓取大小(Setting the fetch size)

抓取大小指的是一次從cassandra獲取到的記錄數(shù)，換句話(huà)說(shuō)，就是每一頁(yè)的記錄數(shù);我們能夠在創(chuàng)建cluster實(shí)例的時(shí)候給它的fetch size指定一個(gè)默認(rèn)值，如果沒(méi)有指定，那么默認(rèn)是5000  

// At initialization: 
Cluster cluster = Cluster.builder() 
    .addContactPoint("127.0.0.1") 
    .withQueryOptions(new QueryOptions().setFetchSize(2000)) 
    .build(); 
 
// Or at runtime: 
cluster.getConfiguration().getQueryOptions().setFetchSize(2000);  

另外，statement上也能設(shè)置fetch size 

Statement statement = new SimpleStatement("your query"); 
 
statement.setFetchSize(2000);  

如果statement上設(shè)置了fetch size，那么statement的fetch size將起作用，否則則是cluster上的fetch size起作用。

注意：設(shè)置了fetch size并不意味著cassandra總是返回準(zhǔn)確的結(jié)果集(等于fetch size)，它可能返回比f(wàn)etch size稍微多一點(diǎn)或者少一點(diǎn)的結(jié)果集。

二、結(jié)果集迭代

fetch size限制了每一頁(yè)返回的結(jié)果集的數(shù)量，如果你迭代某一頁(yè)，驅(qū)動(dòng)會(huì)在后臺(tái)自動(dòng)的抓取下一頁(yè)的記錄。如下例，fetch size = 20：   

默認(rèn)情況下，后臺(tái)自動(dòng)抓取發(fā)生在最后一刻，也就是當(dāng)某一頁(yè)的記錄被迭代完的時(shí)候。如果需要更好的控制，ResultSet接口提供了以下方法：

getAvailableWithoutFetching() and isFullyFetched() to check the current state; 
 
fetchMoreResults() to force a page fetch;  

以下是如何使用這些方法提前預(yù)取下一頁(yè)，以避免在某一頁(yè)迭代完后才抓取下一頁(yè)造成的性能下降：  

ResultSet rs = session.execute("your query"); 
for (Row row : rs) { 
    if (rs.getAvailableWithoutFetching() == 100 && !rs.isFullyFetched()) 
        rs.fetchMoreResults(); // this is asynchronous 
    // Process the row ... 
    System.out.println(row); 
}  

三、保存并重新使用分頁(yè)狀態(tài)

有時(shí)候，將分頁(yè)狀態(tài)保存起來(lái)，對(duì)以后的恢復(fù)是非常有用的，想象一下：有一個(gè)無(wú)狀態(tài)Web服務(wù)，顯示結(jié)果列表，并顯示下一頁(yè)的鏈接，當(dāng)用戶(hù)點(diǎn)擊這個(gè)鏈接的時(shí)候，我們需要執(zhí)行與之前完全相同的查詢(xún)，除了迭代應(yīng)該從上一頁(yè)停止的位置開(kāi)始;相當(dāng)于記住了上一頁(yè)迭代到了哪了，那么下一頁(yè)從這里開(kāi)始即可。

為此，驅(qū)動(dòng)程序會(huì)暴露一個(gè)PagingState對(duì)象，該對(duì)象表示下一頁(yè)被提取時(shí)我們?cè)诮Y(jié)果集中的位置。 

ResultSet resultSet = session.execute("your query"); 
// iterate the result set... 
PagingState pagingState = resultSet.getExecutionInfo().getPagingState(); 
 
// PagingState對(duì)象可以被序列化成字符串或字節(jié)數(shù)組 
String string = pagingState.toString(); 
byte[] bytes = pagingState.toBytes();  

PagingState對(duì)象被序列化后的內(nèi)容可以持久化存儲(chǔ)起來(lái)，也可用作分頁(yè)請(qǐng)求的參數(shù)，以備后續(xù)再次被利用，反序列化成對(duì)象即可： 

PagingState.fromBytes(byte[] bytes); 
 
PagingState.fromString(String str);  

請(qǐng)注意，分頁(yè)狀態(tài)只能使用完全相同的語(yǔ)句重復(fù)使用(相同的查詢(xún)，相同的參數(shù))。而且，它是一個(gè)不透明的值，只是用來(lái)存儲(chǔ)一個(gè)可以被重新使用的狀態(tài)值，如果嘗試修改其內(nèi)容或?qū)⑵涫褂迷诓煌恼Z(yǔ)句上，驅(qū)動(dòng)程序會(huì)拋出錯(cuò)誤。

具體我們來(lái)看下代碼，下例是模擬頁(yè)面分頁(yè)的請(qǐng)求，實(shí)現(xiàn)遍歷teacher表中的全部記錄：

接口：

import java.util.Map; 
 
import com.datastax.driver.core.PagingState; 
 
public interface ICassandraPage 
{ 
    Map<String, Object> page(PagingState pagingState); 
 
}  

主體代碼：

import java.util.ArrayList; 
import java.util.HashMap; 
import java.util.List; 
import java.util.Map; 
 
import com.datastax.driver.core.PagingState; 
import com.datastax.driver.core.ResultSet; 
import com.datastax.driver.core.Row; 
import com.datastax.driver.core.Session; 
import com.datastax.driver.core.SimpleStatement; 
import com.datastax.driver.core.Statement; 
import com.huawei.cassandra.dao.ICassandraPage; 
import com.huawei.cassandra.factory.SessionRepository; 
import com.huawei.cassandra.model.Teacher; 
 
public class CassandraPageDao implements ICassandraPage 
{ 
    private static final Session session = SessionRepository.getSession(); 
     
    private static final String CQL_TEACHER_PAGE = "select * from mycas.teacher;"; 
     
    @Override 
    public Map<String, Object> page(PagingState pagingState) 
    { 
        final int RESULTS_PER_PAGE = 2; 
        Map<String, Object> result = new HashMap<String, Object>(2); 
        List<Teacher> teachers = new ArrayList<Teacher>(RESULTS_PER_PAGE); 
 
        Statement st = new SimpleStatement(CQL_TEACHER_PAGE); 
        st.setFetchSize(RESULTS_PER_PAGE); 
         
        // 第一頁(yè)沒(méi)有分頁(yè)狀態(tài) 
        if (pagingState != null) 
        {             
            st.setPagingState(pagingState); 
        } 
         
        ResultSet rs = session.execute(st); 
        result.put("pagingState", rs.getExecutionInfo().getPagingState()); 
         
        //請(qǐng)注意，我們不依賴(lài)RESULTS_PER_PAGE，因?yàn)?span id="k6zqhab033oa"    class="keyword">fetch size并不意味著cassandra總是返回準(zhǔn)確的結(jié)果集 
        //它可能返回比fetch size稍微多一點(diǎn)或者少一點(diǎn)，另外，我們可能在結(jié)果集的結(jié)尾 
        int remaining = rs.getAvailableWithoutFetching(); 
        for (Row row : rs) 
        { 
            Teacher teacher = this.obtainTeacherFromRow(row); 
            teachers.add(teacher); 
             
            if (--remaining == 0)  
            { 
                break; 
            } 
        } 
        result.put("teachers", teachers); 
        return result; 
    } 
 
    private Teacher obtainTeacherFromRow(Row row) 
    { 
        Teacher teacher = new Teacher(); 
        teacher.setAddress(row.getString("address")); 
        teacher.setAge(row.getInt("age")); 
        teacher.setHeight(row.getInt("height")); 
        teacher.setId(row.getInt("id")); 
        teacher.setName(row.getString("name")); 
         
        return teacher; 
    } 
  
}  

測(cè)試代碼：

import java.util.Map; 
 
import com.datastax.driver.core.PagingState; 
import com.huawei.cassandra.dao.ICassandraPage; 
import com.huawei.cassandra.dao.impl.CassandraPageDao; 
 
public class PagingTest 
{ 
     
    public static void main(String[] args) 
    { 
        ICassandraPage cassPage = new CassandraPageDao(); 
        Map<String, Object> result = cassPage.page(null); 
        PagingState pagingState = (PagingState) result.get("pagingState"); 
        System.out.println(result.get("teachers")); 
        while (pagingState != null) 
        { 
            // PagingState對(duì)象可以被序列化成字符串或字節(jié)數(shù)組 
            System.out.println("=============================================="); 
            result = cassPage.page(pagingState); 
            pagingState = (PagingState) result.get("pagingState"); 
            System.out.println(result.get("teachers")); 
        } 
    } 
     
}  

我們來(lái)看看Statement的setPagingState(pagingState)方法： 

四、偏移查詢(xún)

保存分頁(yè)狀態(tài)，能夠保證從某一頁(yè)移動(dòng)到下一頁(yè)很好地運(yùn)行(也可以實(shí)現(xiàn)上一頁(yè))，但是它不滿(mǎn)足隨機(jī)跳躍，比如直接跳到第10頁(yè)，因?yàn)槲覀儾恢赖?0頁(yè)的前一頁(yè)的分頁(yè)狀態(tài)。像這樣需要偏移查詢(xún)的特點(diǎn)，并不被cassandra原生支持，理由是偏移查詢(xún)效率低下(性能與跳過(guò)的行數(shù)呈線(xiàn)性反比)，所以cassandra官方不鼓勵(lì)使用偏移量。如果非要實(shí)現(xiàn)偏移查詢(xún)，我們可以在客戶(hù)端模擬實(shí)現(xiàn)。但是性能還是呈線(xiàn)性反比，也就說(shuō)偏移量越大，性能越低，如果性能在我們的接受范圍內(nèi)，那還是可以實(shí)現(xiàn)的。例如，每一頁(yè)顯示10行，最多顯示20頁(yè)，這就意味著，當(dāng)顯示第20頁(yè)的時(shí)候，最多需要額外的多抓取190行，但這也不會(huì)對(duì)性能造成太大的降低，所以數(shù)據(jù)量不大的話(huà)，模擬實(shí)現(xiàn)偏移查詢(xún)還是可以的。

舉個(gè)例子，假設(shè)每頁(yè)顯示10條記錄，fetch size 是50，我們請(qǐng)求第12頁(yè)(也就是第110行到第119行)：

1、第一次執(zhí)行查詢(xún)，結(jié)果集包含0到49行，我們不需要用到它，只需要分頁(yè)狀態(tài);

2、用第一次查詢(xún)得到的分頁(yè)狀態(tài)，執(zhí)行第二次查詢(xún);

3、用第二次查詢(xún)得到的分頁(yè)狀態(tài)，執(zhí)行第三次查詢(xún)。結(jié)果集包含100到149行;

4、用第三次查詢(xún)得到的結(jié)果集，先過(guò)濾掉前10條記錄，然后讀取10條記錄，最后丟棄剩下的記錄，讀取的10條記錄則是第12頁(yè)需要顯示的記錄。

我們需要嘗試著找到最佳的fetch size來(lái)達(dá)到最佳平衡：太小就意味著后臺(tái)更多的查詢(xún);太大則意味著返回了更大的信息量以及更多不需要的行。

另外，cassandra本身不支持偏移量查詢(xún)。在滿(mǎn)足性能的前提下，客戶(hù)端模擬偏移量的實(shí)現(xiàn)只是一種妥協(xié)。官方建議如下：

1、使用預(yù)期的查詢(xún)模式來(lái)測(cè)試代碼，以確保假設(shè)是正確的

2、設(shè)置最高頁(yè)碼的硬限制，以防止惡意用戶(hù)觸發(fā)跳過(guò)大量行的查詢(xún)

五、總結(jié)

Cassandra對(duì)分頁(yè)的支持有限，上一頁(yè)、下一頁(yè)比較好實(shí)現(xiàn)。不支持偏移量的查詢(xún)，硬要實(shí)現(xiàn)的話(huà)，可以采用客戶(hù)端模擬的方式，但是這種場(chǎng)景最好不要用在cassandra上，因?yàn)閏assandra一般而言是用來(lái)解決大數(shù)據(jù)問(wèn)題，而偏移量查詢(xún)一旦數(shù)據(jù)量太大，性能就不敢恭維了。

在我的項(xiàng)目中，索引修復(fù)用到了cassandra的分頁(yè)，場(chǎng)景如下：cassandra的表不建二級(jí)索引，用elasticsearch實(shí)現(xiàn)cassandra表的二級(jí)索引，那么就會(huì)涉及到索引的一致性修復(fù)的問(wèn)題，這里就用到了cassandra的分頁(yè)，對(duì)cassandra的某張表進(jìn)行全表遍歷，逐條與elasticsearch中的數(shù)據(jù)進(jìn)行匹對(duì)，若elasticsearch中不存在，則在elasticsearch中新增，若存在而又不一致，則在elasticsearch中修復(fù)。具體elasticsearch怎么樣實(shí)現(xiàn)cassandra的索引功能，在我后續(xù)博客中會(huì)專(zhuān)門(mén)的講解，這里就不多說(shuō)了。而在cassandra表進(jìn)行全表遍歷的時(shí)候就需要用到分頁(yè)，因?yàn)楸碇袛?shù)據(jù)量太大，億級(jí)別的數(shù)據(jù)不可能一次全部加載到內(nèi)存中。