封装好的MAP工具类和HBASE工具类

package cn.gulf.util;

import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import javax.servlet.http.HttpServletRequest;

public class PageData extends HashMap implements Map{
	
	private static final long serialVersionUID = 1L;
	
	Map map = null;
	HttpServletRequest request;
	
	public PageData(HttpServletRequest request){
		this.request = request;
		Map properties = request.getParameterMap();
		Map returnMap = new HashMap(); 
		Iterator entries = properties.entrySet().iterator(); 
		Map.Entry entry; 
		String name = "";  
		String value = "";  
		while (entries.hasNext()) {
			entry = (Map.Entry) entries.next(); 
			name = (String) entry.getKey(); 
			Object valueObj = entry.getValue(); 
			if(null == valueObj){ 
				value = ""; 
			}else if(valueObj instanceof String[]){ 
				String[] values = (String[])valueObj;
				for(int i=0;i<values.length;i++){ 
					 value = values[i] + ",";
				}
				value = value.substring(0, value.length()-1); 
			}else{
				value = valueObj.toString(); 
			}
			returnMap.put(name, value); 
		}
		map = returnMap;
	}
	
	public PageData() {
		map = new HashMap();
	}
	
	@Override
	public Object get(Object key) {
		Object obj = null;
		if(map.get(key) instanceof Object[]) {
			Object[] arr = (Object[])map.get(key);
			obj = request == null ? arr:(request.getParameter((String)key) == null ? arr:arr[0]);
		} else {
			obj = map.get(key);
		}
		return obj;
	}
	
	public String getString(Object key) {
		return (String)get(key);
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public Object put(Object key, Object value) {
		return map.put(key, value);
	}
	
	@Override
	public Object remove(Object key) {
		return map.remove(key);
	}

	public void clear() {
		map.clear();
	}

	public boolean containsKey(Object key) {
		// TODO Auto-generated method stub
		return map.containsKey(key);
	}

	public boolean containsValue(Object value) {
		// TODO Auto-generated method stub
		return map.containsValue(value);
	}

	public Set entrySet() {
		// TODO Auto-generated method stub
		return map.entrySet();
	}

	public boolean isEmpty() {
		// TODO Auto-generated method stub
		return map.isEmpty();
	}

	public Set keySet() {
		// TODO Auto-generated method stub
		return map.keySet();
	}

	@SuppressWarnings("unchecked")
	public void putAll(Map t) {
		// TODO Auto-generated method stub
		map.putAll(t);
	}

	public int size() {
		// TODO Auto-generated method stub
		return map.size();
	}

	public Collection values() {
		// TODO Auto-generated method stub
		return map.values();
	}
	
}

package cn.gulf.repository;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
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.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

import org.apache.log4j.Logger;
import org.springframework.beans.BeanWrapper;
import org.springframework.beans.BeanWrapperImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Repository;

import cn.gulf.PageData;

@Repository
public class HbaseQueryDao {

	Logger logger = Logger.getLogger(HbaseQueryDao.class); 
	
	private String columnFamily = "columns";

	public String getColumnFamily() {
		return columnFamily;
	}

	public void setColumnFamily(String columnFamily) {
		this.columnFamily = columnFamily;
	}

	@Autowired
	private HbaseTemplate hbaseTemplate;

	public <T> T queryForBeanByRowKey(String tableName, String rowKey, final Class<T> beanType) {
		
		logger.info("-----------------------HbaseQueryDao.queryForBeanByRowKey-----------------------------------");
		logger.info("tableName:" + tableName + ",rowKey:" + rowKey);
		logger.info("-----------------------HbaseQueryDao.queryForBeanByRowKey-----------------------------------");
        return hbaseTemplate.get(tableName, rowKey, new RowMapper<T>() {  
            public T mapRow(Result result, int rowNum) throws Exception {  

            	Map<byte[], byte[]> map = result.getFamilyMap(Bytes.toBytes(columnFamily));
            	T t = beanType.newInstance();
            	BeanWrapper beanWrapper = new BeanWrapperImpl(t);
                for(Map.Entry<byte[], byte[]> entry : map.entrySet()){
                	beanWrapper.setPropertyValue(Bytes.toString(entry.getKey()),Bytes.toString(entry.getValue()));
                }
                return t;  
            }  
        });  
	}
	
	/**
	 * 根据精确的rowKey查询
	 * @param tableName
	 * @param rowKey
	 * @return
	 */
	public Map<String, String> queryForMapByRowKey(String tableName, String rowKey) {
		
		logger.info("-----------------------HbaseQueryDao.queryForMapByRowKey-----------------------------------");
		logger.info("tableName:" + tableName + ",rowKey:" + rowKey);
		logger.info("-----------------------HbaseQueryDao.queryForMapByRowKey-----------------------------------");
        return hbaseTemplate.get(tableName, rowKey, new RowMapper<Map<String, String>>() {  
            public Map<String, String> mapRow(Result result, int rowNum) throws Exception {  

            	Map<byte[], byte[]> mapColumn = result.getFamilyMap(Bytes.toBytes(columnFamily));
            	Map<String, String> map = new HashMap<String, String>();
                for(Map.Entry<byte[], byte[]> entry : mapColumn.entrySet()){
                	map.put(Bytes.toString(entry.getKey()),Bytes.toString(entry.getValue()));
                }

                return map;  
            }  
        });  
	}
	
   /**
    * 根据startRow列表
    * @param tableName
    * @param startRow
    * @param stopRow
    * @return
    */
	public List<PageData> queryForListByScanRange(String tableName, String startRow, String stopRow) {
		
		logger.info("-----------------------HbaseQueryDao.queryForListByScanRange-----------------------------------");
		logger.info("tableName:" + tableName + ",startRow:" + startRow+ ",stopRow:" + stopRow);
		logger.info("-----------------------HbaseQueryDao.queryForListByScanRange-----------------------------------");
		
		Scan scan = new Scan();  
		if (startRow != null) {
			scan.setStartRow(Bytes.toBytes(startRow));
        } else {
        	scan.setStartRow(Bytes.toBytes(""));
        }
        if (stopRow != null) {
        	scan.setStopRow(Bytes.toBytes(stopRow));
        } else {
        	scan.setStopRow(Bytes.toBytes(""));
        }
        
        Filter pf = new PrefixFilter(Bytes.toBytes(startRow));
        scan.setFilter(pf);

        return hbaseTemplate.find(tableName, scan, new RowMapper<PageData>() {
			
            public PageData mapRow(Result result, int rowNum) throws Exception {  

            	Map<byte[], byte[]> mapColumn = result.getFamilyMap(Bytes.toBytes(columnFamily));
            	PageData map = new PageData();
                for(Map.Entry<byte[], byte[]> entry : mapColumn.entrySet()){
                	map.put(Bytes.toString(entry.getKey()), Bytes.toString(entry.getValue()));
                }
                return map;  
            }  
        });  
	}

	public void save(String tableName, String rowKey, Map<String,String> rowMap) {

		for (Entry<String, String> en : rowMap.entrySet()) {
			hbaseTemplate.put(tableName, rowKey, columnFamily, en.getKey(), Bytes.toBytes(en.getValue()));
		}
	}
	
	public void delete(String tableName, final String rowKey) {
		
		  hbaseTemplate.execute(tableName, new TableCallback<Boolean>() {  
		        public Boolean doInTable(HTableInterface table) throws Throwable {  
		            boolean flag = false;  
		            try{  
		                List<Delete> list = new ArrayList<Delete>();  
		                Delete d1 = new Delete(rowKey.getBytes());  
		                list.add(d1);  
		            	table.delete(list); 
		             flag = true;  
		            }catch(Exception e){  
		                e.printStackTrace();  
		            }  
		            return flag;  
		        }  
		    });  
	}
	
}