indexserver.java

nutch搜索的改进型工具和优化爬虫的相关工具

/*
 * 创建日期 2005-9-20
 *
 * TODO 要更改此生成的文件的模板，请转至
 * 窗口 － 首选项 － Java － 代码样式 － 代码模板
 */
package net.nutch.searcher;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;

/**
 * @author Administrator
 *
 * TODO 要更改此生成的类型注释的模板，请转至
 * 窗口 － 首选项 － Java － 代码样式 － 代码模板
 */
public class IndexServer {
	
	public static final long RESTORE_SCALE_MILLISECOND = 20000; //10 second
	
	public static final Logger LOG = Logger.getLogger("timeout");
	
  	public long[] maxDocs;
	
  	private String flag="";
	
	private ArrayList<InetSocketAddress[]> addresses = 
		new ArrayList<InetSocketAddress[]>();
		
	private ConcurrentHashMap<String,Integer> segmentToAddMap = 
		new ConcurrentHashMap<String,Integer>();

	public static class IndexInetSocketAddress extends InetSocketAddress{
		public static final int MAX_TIMEOUT_LEVEL = 8;
		private int timeout_level = 0;
		private long timeout_time = 0;
		private boolean online = true;

		public IndexInetSocketAddress(int port) {
			super(port);
		}
		
		public IndexInetSocketAddress(InetAddress addr, int port){
			super(addr, port);
		}
		
		public IndexInetSocketAddress(String hostname, int port){
			super(hostname, port);
		}
		
		public IndexInetSocketAddress(InetSocketAddress addr){
			super(addr.getAddress() , addr.getPort() );
		}

		public void setOnline(boolean online) {
			this.online = online;
		}
		
		public boolean isOnline() {
			return this.online;
		}
		
		public int getTimeoutLevel() {
			return this.timeout_level;
		}
		
		public void resetTimeout(){
			this.timeout_level = 0;
			this.timeout_time = System.currentTimeMillis();
		}
		
		public void incTimeout(){
			if (this.timeout_level < MAX_TIMEOUT_LEVEL)
				this.timeout_level ++;
			this.timeout_time = System.currentTimeMillis();
		}
		
		public boolean isValid(long restoreScaleMilliSecond){
			if (this.online){
				if (this.timeout_level > 0){
					if (System.currentTimeMillis() >= 
						this.timeout_time + this.timeout_level *restoreScaleMilliSecond)
						return true;
					return false;
				}
				return true;
			}
			return false;
		}
		
		public String toString() {
			String value;
			value = super.toString() + "\nTimeout_Level: --- " + timeout_level +
				 " --- Timeout_Time: " + timeout_time;
			return value;
		}
	}
	
	public IndexServer(){
	}
	
	public void init() {
		
	}
	
  	/**
  	 * Overloaded addSegment(String, InetSocketAddress
  	 * )
  	 * Added by dingzhenbo
  	 * @param segment
  	 * @param addr1
  	 */
	public void addSegment(String segment, InetSocketAddress addr1) {
		for(int i=0; i<addresses.size(); i++){
			InetSocketAddress[] address = addresses.get(i);
			for (InetSocketAddress addr : address){
				if (addr.getAddress().getHostAddress().equals(addr1.getAddress().getHostAddress()) &&
						addr.getPort() == addr1.getPort()) {
					segmentToAddMap.put(segment,i);
					return;
				}
			}
		}		
	}
	
	public void addSegment(String segment, InetSocketAddress[] addrs){
		for(int i=0; i<addresses.size(); i++){
			InetSocketAddress[] address = addresses.get(i);
			for (InetSocketAddress addr : address){
				for (InetSocketAddress addr1 : addrs){
					if (addr.getAddress().getHostAddress().equals(addr1.getAddress().getHostAddress()) &&
							addr.getPort() == addr1.getPort()){
						segmentToAddMap.put(segment,i);
						return;
					}
				}
			}
		}
	}
	
	public void delSegment(String segment){
		segmentToAddMap.remove(segment);
	}
	
	public void clear(){
		segmentToAddMap.clear();
	}
	
	public InetSocketAddress[] getSegment(String segment){
		Integer index = (Integer)segmentToAddMap.get(segment);
		if (index == null)
			return null;
		return get(index.intValue());
	}
	
	public String[] getSegmentNames(){
		return (String[])segmentToAddMap.keySet().toArray(new String[segmentToAddMap.size()]);
	}
	
	public InetSocketAddress getAddr(String host, int port){
		for( InetSocketAddress[] address : addresses){
			for( InetSocketAddress addr : address){
				if (addr.getAddress().getHostAddress().equals(host) && addr.getPort() == port)
					return addr;
			}
		}
		return null;
	}
	
	public void add(InetSocketAddress[] address){
		if (address == null || address.length ==0 )
			return;
		int i;
		IndexInetSocketAddress[] newAddr = new IndexInetSocketAddress[address.length];
		for (i=0; i<address.length; i++){
			newAddr[i] = new IndexInetSocketAddress(address[i]);
		}
		addresses.add(newAddr);
	}
	
	public InetSocketAddress[] get(int i){
		return (InetSocketAddress[])addresses.get(i);
	}
	
	public int getLength(){
		return addresses.size();
	}
	
	public boolean getOnlineStat(InetSocketAddress addr){
		if (addr == null)
			return false;
		if (addr instanceof IndexInetSocketAddress)
			return ((IndexInetSocketAddress)addr).isValid(RESTORE_SCALE_MILLISECOND);
		IndexInetSocketAddress value = 
			(IndexInetSocketAddress)getAddr(addr.getAddress().getHostAddress(),addr.getPort());
		if (value == null)
			return false;
		return value.isValid(RESTORE_SCALE_MILLISECOND);
	}
	
	public void setOnlineStat(InetSocketAddress addr, boolean stat){
		if (addr == null)
			return;
		if (addr instanceof IndexInetSocketAddress)
			((IndexInetSocketAddress)addr).setOnline(stat);
		else{
			IndexInetSocketAddress value = 
				(IndexInetSocketAddress)getAddr(addr.getAddress().getHostAddress(),addr.getPort());
			if (value == null)
				return;
			value.setOnline(stat);
		}
	}
	
	public InetSocketAddress select(InetSocketAddress[] indexs){
    	ArrayList<Integer> onLine = new ArrayList<Integer>();
		for (int j=0; j<indexs.length; j++){
			if (getOnlineStat(indexs[j])){
				onLine.add(j);
			}
		}
		if (onLine.size() == 0){
			LOG.warn("Index server total " + indexs.length + " hosts, but all server offline! One host is: " +
					indexs[0].getAddress().getHostAddress() + ":" + indexs[0].getPort());
			return null;
		}
		int index = onLine.get(0).intValue();
		if (onLine.size() > 1){
			Random r = new Random(System.currentTimeMillis());
			index = r.nextInt(1000) % onLine.size();
			index = onLine.get(index).intValue();
		}
		return indexs[index];
    }
	
	public InetSocketAddress[] select(){
		ArrayList<InetSocketAddress> addrs = new ArrayList<InetSocketAddress>();
		for (int i=0; i<addresses.size(); i++){
			InetSocketAddress addr = select((InetSocketAddress[])addresses.get(i));
			//if (addr == null)
			//	continue;
			addrs.add(addr);
		}
		if (addrs.size()==0){
			return null;
		}
		return (InetSocketAddress[])addrs.toArray(new InetSocketAddress[addrs.size()]);
	}
	
	public boolean contains(InetSocketAddress inetAddr) {
		for (int i=0; i<addresses.size(); i++) {
			InetSocketAddress[] addrs = addresses.get(i);
			for (InetSocketAddress addr : addrs) {
				if (addr.equals(inetAddr)) {
					return true;
				}
			}
		}
		return false;
	}

	public String getModeFlag() {
		return flag;
	}

	public void setModeFlag(String flag) {
		this.flag = flag;
	}
}