基于HBase的冠字号查询系统2--实现部分

1)        根据给定的取款冠字号个数num，随机查找冠字号（rowkey）对应的op_www:exist字段值为1的num*3条记录；

2)        使用HBase.checkAndPut进行更新，把op_www:exist字段值更新为0，并返回更新后的rowkey，即冠字号；

3)        如果在num*3条记录更新后，被更新的冠字号不足num条，则再次随机查找冠字号对应的op_www:exist字段值为1的记录，并更新，返回更新后的冠字号，直到返回的冠字号个数为num；

package stumer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

public class ReadTest {
	
//	private static String FAMILY ="info"; 
	
	
	public static void main(String[] args) throws IOException {
		long size =10000;
		get(Utils.getConn(),Utils.generateRowKey(size));
	}
	
	public static void get(Connection connection,List<byte[]> rowkeys) throws IOException {
		System.out.println(new Date()+":开始读取记录...");
		long start =System.currentTimeMillis();
		Table table = connection.getTable(TableName.valueOf(Utils.TABLE));
		Get get = null ;
		long count =0;
		try{
			for(byte[] rowkey :rowkeys){
				count ++;
	//			get = new Get(Bytes.toBytes(""));
				get = new Get(rowkey);
				table.get(get);
				if(count%1000==0){
					System.out.println("count:"+count);
				}
			}
		long end = System.currentTimeMillis();
		System.out.println(new Date()+":"+rowkeys.size()+"条记录，读取耗时:"+(end-start)*1.0/1000+"s");
		}catch(Exception e){
			
		}finally{
			table.close();
		}
		
	}
	
	
}

package stumer;

import java.util.Date;
import java.util.List;

import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Table;

public class ReadThread implements Runnable {

	private List<byte[]> rks;
	private Table table;
	public  ReadThread(Table table ,List<byte[]> rks) {
		this.table = table;
		this.rks = rks;
	}
	@Override
	public void run() {
		System.out.println(Thread.currentThread().getName()+" "+new Date()+":开始读取记录...");
		long start =System.currentTimeMillis();
		Get get = null ;
		long count =0;
		try{
			for(byte[] rowkey :rks){
				count ++;
	//			get = new Get(Bytes.toBytes(""));
				get = new Get(rowkey);
				table.get(get);
				if(count%1000==0){
					System.out.println(Thread.currentThread().getName()+" count:"+count);
				}
			}
		long end = System.currentTimeMillis();
		System.out.println(Thread.currentThread().getName()+" "+new Date()
				+":"+rks.size()+"条记录，读取耗时:"+(end-start)*1.0/1000+"s");
		}catch(Exception e){
			
		}
		
	}

}

package stumer;

import java.io.IOException;

public class ReadThreadTest {

	public static void main(String[] args) throws IOException {
		long dataSize =500;
		int threadSize = 20;
		for(int i=0;i<threadSize;i++){
			new Thread(new ReadThread(Utils.getTable(), Utils.generateRowKey(dataSize))).start();
		}
	}
}

package stumer;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

public class Utils {
	public static String TABLE = "records";
	private static DecimalFormat df = new DecimalFormat( "0000" );
	public static String[]  crownSizePrefixes =null;
	static Random random = new Random();

	static {
		crownSizePrefixes = new String[26*2];
		for (int i = 0; i < crownSizePrefixes.length/2; i++) {
			crownSizePrefixes[i] = "AAA" + (char) (65 + i);
			crownSizePrefixes[i+26] = "AAB" + (char) (65 + i);		
		}
	}
	/**
	 * 把0~9999 转为 0000~9999
	 * @param num
	 * @return
	 */
	public static String formatCrownSizeSuffix(int num){
		return df.format(num);
	}
	public static Table getTable() throws IOException{
		return getConn().getTable(TableName.valueOf(TABLE));
	}
	public static String getRandomCrownSize(){
		return crownSizePrefixes[random.nextInt(crownSizePrefixes.length)]
				+formatCrownSizeSuffix(random.nextInt(10000));
	}
	public static Connection getConn() throws IOException {
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.master", "node2:16000");// 指定HMaster
		conf.set("hbase.rootdir", "hdfs://node1:8020/hbase");// 指定HBase在HDFS上存储路径
		conf.set("hbase.zookeeper.quorum", "node2,node3,node4");// 指定使用的Zookeeper集群
		conf.set("hbase.zookeeper.property.clientPort", "2181");// 指定使用Zookeeper集群的端口

		Connection connection = ConnectionFactory.createConnection(conf);// 获取连
		return connection;
	}
	
	public static List<byte[]> generateRowKey(long size){
		System.out.println(new Date()+"开始生成"+size +"条记录...");
		long start =System.currentTimeMillis();
		List<byte[]>  rowkeys = new ArrayList<>();
		
		for(int i=0;i<size;i++){
			rowkeys.add(Bytes.toBytes(Utils.getRandomCrownSize()));
		}
		long end =System.currentTimeMillis();
		System.out.println(new Date()+":"+rowkeys.size()+"条记录，生成耗时:"+(end-start)*1.0/1000+"s");
		return rowkeys;
	}
}

3. 总结

分享，成长，快乐

脚踏实地，专注

转载请注明blog地址：http://blog.csdn.net/fansy1990