hbase 分页代码

待我称王封你为后i 2022-08-07 08:54 205阅读 0赞

众所周知，hbase通过scan来扫描表，通过startKey，stopKey来确定范围，hbase官方提供了一个PageFilter来支持一次scan可以返回多少条数据即每页的行数。假如一页是10条，这样是第一页还好，但是第二页呢，如果不改变PageFilter的pageSize，那返回的还是第一页的数据，如果改变pageSize为20，则返回了第一页10多余的数据，在客户端要过滤掉，性能不好。那怎么办呢，方法就是在查询下一页时，指定下一页的startKey，这样PageFilter每次就不会返回多余的记录，stopKey可以不用变，那现在问题是，怎么得到下一页的startKey(即下一页第一行的rowkey)呢?,有两种方法来取每一页的startKey  
  
一  上一页的最后一行记录的rowkey作为下一页的startKey。  
二   在每次scan时多取一条记录，即把下一页第一条行页取出来，把该行的rowkey做为下一页的startKey。  
  
不管用一还是二，都要注意，hbase scan时是包含startKey的，如果是采用第一种，则要在记录多取一条，排除第一条。第二种页是多取一条，但是排除最后一条，用来做下一页的startKey。还有需要注意的是在计算是否有下一页时，可以根据返回的条数来判断。  
  
startKey怎么取没有问题了。但是怎么存储呢，有同学可能会想到存到session，但是如果你的服务是rest api型的，就没有session的概念了。那还有两种选择：  
一 是存到客户端，让客户端每次请求时把startKey再传回来，这样需要依赖客户端，如果客户端是远程，或者是开放平台的情况下，可能不合适。  
二 存在服务端，存在服务端需要注意并发访问的情况。比如scan同一个表，一个访问第2页，一个访问第3页，服务端就需要对每一个table的scan 存每一页的startKey，需要为同一个查询条件包含pageSize，因为pageSize不一样，startKey也会不一样，  
在服务crash情况下，从起后都从第一页开始。  
  
我自己是采用第二种方案，存在服务端，需要代码的，如果你有更好的方案的请分享出来。

注：hbase的分页一般只提供下一页，不提供直接最后一页，

代码如下：

import java.util.Iterator;  
import java.util.Map;  
import java.util.concurrent.ConcurrentHashMap;  
  
/\*\*  
 \* 分页监控器  
 \*  
 \* @author bahaidong  
 \*  
 \*/  
public class PageScanManager \{  
    //失效时间10分钟  
    private final static long expirtTime = 10 \* 60 \* 1000;  
    private Map<String, PageScan> scanMaps = new ConcurrentHashMap<String, PageScan>();  
    private static PageScanManager instance = new PageScanManager();  
  
    private PageScanManager() \{  
  
        Thread monitor = new Thread(new ScanMonitor(),  
                "pagescan-monitor-thread");  
        monitor.setDaemon(true);  
        monitor.start();  
    \}  
  
    public static PageScanManager getInstance() \{  
        return instance;  
    \}  
  
    public PageScan getPageScan(String scanStartKey, String scanEndKey,  
            int pageSize) \{  
        String scanKey = scanStartKey + scanEndKey + pageSize;  
        PageScan ps = scanMaps.get(scanKey);  
        if (ps != null)  
            return ps;  
        synchronized (scanMaps) \{  
            if (scanMaps.get(scanKey) == null) \{  
                PageScan nps = new PageScan(scanStartKey, scanEndKey, pageSize);  
                scanMaps.put(nps.getKEY(), nps);  
            \}  
        \}  
        return scanMaps.get(scanKey);  
    \}  
  
    /\*\*  
     \* 内部监控PageScan的线程，如果某个PageScan在指定的时间内没有被访问 。则销毁  
     \*  
     \* @author bahaidong  
     \*  
     \*/  
    private class ScanMonitor implements Runnable \{  
  
        @Override  
        public void run() \{  
            while (true) \{  
                long now = System.currentTimeMillis();  
                Iterator<PageScan> it = scanMaps.values().iterator();  
                while (it.hasNext()) \{  
                    PageScan pS = it.next();  
                    if ((now - expirtTime) > pS.getStartTime()) \{  
                        String k = pS.getKEY();  
                        scanMaps.remove(k);  
                    \}  
                \}  
                try \{  
                    Thread.sleep(5000);  
                \} catch (InterruptedException e) \{  
                    e.printStackTrace();  
                \}  
  
            \}  
        \}  
  
    \}  
  
\}

import java.util.HashMap;  
import java.util.Map;  
  
/\*\*  
 \* hbase 分页  
 \*  
 \* @author bahaidong  
 \*  
 \*/  
public class PageScan \{  
  
    private String sKey;  
    private String eKey;  
    private int pageSize;  
    private long startTime;  
    private Map<Integer, String> pageRowKey;  
  
    public PageScan(String startKey, String endKey, int pageSize) \{  
        this.sKey = startKey;  
        this.eKey = endKey;  
        this.pageSize = pageSize;  
        this.startTime = System.currentTimeMillis();  
    \}  
  
    public synchronized void setPageNoStartKey(int pageNo, String startKey) \{  
        if (pageRowKey == null) \{  
            pageRowKey = new HashMap<Integer, String>();  
        \}  
        String prePageRowKey = pageRowKey.get(pageNo);  
        if (prePageRowKey != null) \{  
            pageRowKey.remove(pageNo);  
        \}  
        pageRowKey.put(pageNo, startKey);  
    \}  
  
    public String getPageStartKey(int pageNo) throws Exception \{  
        if (pageNo <= 1) \{  
            return this.sKey;  
        \}  
        this.startTime = System.currentTimeMillis();  
        if (this.pageRowKey == null) \{  
            throw new Exception(  
                    " pageNo must start with 1 when useHasNext is true ");  
        \}  
        String nextK = this.pageRowKey.get(pageNo);  
        if (nextK == null) \{  
            return this.sKey;  
        \}  
        return nextK;  
    \}  
  
    public String getKEY() \{  
        return this.sKey + this.eKey + this.pageSize;  
    \}  
  
    public String getsKey() \{  
        return sKey;  
    \}  
  
    public void setsKey(String sKey) \{  
        this.sKey = sKey;  
    \}  
  
    public String geteKey() \{  
        return eKey;  
    \}  
  
    public void seteKey(String eKey) \{  
        this.eKey = eKey;  
    \}  
  
    public long getStartTime() \{  
        return startTime;  
    \}  
  
    public void setStartTime(long startTime) \{  
        this.startTime = startTime;  
    \}  
  
    @Override  
    public boolean equals(Object obj) \{  
        if (obj == null)  
            return false;  
        PageScan pre = (PageScan) obj;  
        if (this.sKey.equals(pre.getsKey()) && this.eKey.equals(pre.geteKey())  
                && this.pageSize == pre.pageSize)  
            return true;  
        return false;  
    \}  
  
    @Override  
    public int hashCode() \{  
  
        return new StringBuilder(this.sKey).append(this.eKey)  
                .append(this.pageSize).toString().hashCode();  
    \}  
  
    public int getPageSize() \{  
        return pageSize;  
    \}  
  
    public void setPageSize(int pageSize) \{  
        this.pageSize = pageSize;  
    \}  
  
\}

import java.util.HashMap;  
import java.util.List;  
import java.util.Map;  
  
import org.apache.hadoop.hbase.client.Result;  
import org.apache.hadoop.hbase.client.Scan;  
import org.apache.hadoop.hbase.filter.Filter;  
import org.apache.hadoop.hbase.filter.PageFilter;  
import org.apache.hadoop.hbase.util.Bytes;  
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.data.hadoop.hbase.HbaseTemplate;  
import org.springframework.data.hadoop.hbase.RowMapper;  
import org.springframework.stereotype.Service;  
  
/\*\*  
 \* 分页工具类  
 \* @author bahaidong  
 \*  
 \*/  
@Service  
public class HbasePageUtil \{  
      
    private @Autowired HbaseTemplate template;  
      
    /\*\*  
     \* 返回Map结果集格式为：<br>  
     \* key:hasNext        value:boolean    是否有下一页:true 有， false 没有<br>  
     \* key:results        value:List<Result>    查询结果集<br>  
     \*  
     \* @param tableName  
     \* @param startKey  
     \* @param endKey  
     \* @param pageSize  
     \* @param pageNo  
     \* @return Map<String, Object>  
     \* @throws Exception  
     \*/  
    public Map<String, Object> getResult(String tableName, String startKey,  
            String endKey, int pageSize, int pageNo) throws Exception \{  
        long totalNums = 0;  
        boolean hasNext = false;  
        Map<String, Object> mapResult = new HashMap<String, Object>();  
        PageScanManager psm = PageScanManager.getInstance();  
        PageScan pS = null;  
        if (pageNo >= 1) \{  
            pS = psm.getPageScan(startKey, endKey, pageSize);  
            startKey = pS.getPageStartKey(pageNo);  
        \}  
        Scan scan = new Scan();  
        //scan.setMaxVersions(1);  
        scan.setStartRow(Bytes.toBytes(startKey));  
        scan.setStopRow(Bytes.toBytes(endKey));  
        Filter filter = new PageFilter(pageSize + 1);  
        scan.setFilter(filter);  
          
        List<Result> results = template.find(tableName, scan,  
                new RowMapper<Result>() \{  
                    @Override  
                    public Result mapRow(Result result, int rowNum) throws Exception \{  
                        return result;  
                    \}  
                \});  
        Result lastR = null;  
        if(results != null) \{  
            for(Result s: results) \{  
                totalNums++;  
                lastR = s;  
            \}  
            //默认查询结果不包含endKey，现修改为包含endKey  
            if(totalNums <= pageSize) \{  
                Result result = template.get(tableName, endKey, new RowMapper<Result>()\{  
                    @Override  
                    public Result mapRow(Result result, int rowNum) throws Exception \{  
                        return result;  
                    \}          
                \});  
                if(result != null)\{      
                    //当totalNums==pageSize时有可能为最后一页，最后一页时排除totalNums++的情况，否则可能造成无限循环  
                    if(totalNums < pageSize) \{  
                        totalNums++;  
                        if(!endKey.equals(startKey)) \{  
                            results.add(result);  
                        \}                          
                    \}else \{  
                        lastR = result;  
                        hasNext = true;  
                        String row = Bytes.toString(lastR.getRow());  
                        pS.setPageNoStartKey(pageNo + 1, row);  
                        if(endKey.equals(startKey))\{  
                            hasNext = false;  
                            //results = null;  
                        \}              
                    \}              
                \}  
            \}  
            if(pS != null && totalNums == pageSize + 1) \{  
                results.remove(pageSize);  
                hasNext = true;  
                String row = Bytes.toString(lastR.getRow());  
                pS.setPageNoStartKey(pageNo + 1, row);  
            \}  
        \}  
        mapResult.put("results", results);  
        mapResult.put("hasNext", hasNext);  
          
        return mapResult;  
    \}  
  
\}