java实现memcache服务器的示例代码

VALUE <key> <flags> <bytes>/r/n<data block>/r/nVALUE <key> <flags> <bytes>/r/n<data block>/r/nENDdel

del <key> [noreply]/r/n

 /** * 各种指令 * 目前支持get，set，delete * * 以及自定义的 * error，end */public interface Command {  /**   * 执行指令   * @param reader   * @param writer   */  void execute(Reader reader, Writer writer);  /**   * 获取指令的类型   * @return   */  CommandType getType();}

/** * 指令工厂 单一实例 */public class CommandFactory {  private static CommandFactory commandFactory;  private static Cache<Item> memcache;  private CommandFactory(){}  public static CommandFactory getInstance(Cache<Item> cache) {    if (commandFactory == null) {      commandFactory = new CommandFactory();      memcache = cache;    }    return commandFactory;  }  /**   * 根据指令的类型获取Command   * @param commandLine   * @return   */  public Command getCommand(String commandLine){    if (commandLine.matches("^set .*$")){      return new SetCommand(commandLine, memcache);    }else if (commandLine.matches("^get .*$")){      return new GetCommand(commandLine, memcache);    }else if (commandLine.matches("^del .*$")){      return new DeleteCommand(commandLine, memcache);    }else if (commandLine.matches("^end$")){      return new EndCommand(commandLine);    }else{      return new ErrorCommand(commandLine, ErrorCommand.ErrorType.ERROR);    }  }}

/** * 删除缓存指令 */public class DeleteCommand implements Command{  private final String command;  private final Cache<Item> cache;  private String key;  private boolean noReply;  public DeleteCommand(final String command, final Cache<Item> cache){    this.command = command;    this.cache = cache;    initCommand();  }  private void initCommand(){    if (this.command.contains("noreply")){      noReply = true;    }    String[] info = command.split(" ");    key = info[1];  }  @Override  public void execute(Reader reader, Writer writer) {    BufferedWriter bfw = (BufferedWriter) writer;    Item item = cache.delete(key);    if (!noReply){      try {        if (item == null){          bfw.write("NOT_FOUND/r/n");        }else {          bfw.write("DELETED/r/n");        }        bfw.flush();      } catch (IOException e) {        try {          bfw.write("ERROR/r/n");          bfw.flush();        } catch (IOException e1) {          e1.printStackTrace();        }        e.printStackTrace();      }    }  }  @Override  public CommandType getType() {    return CommandType.SEARCH;  }}

public class Memcache implements Cache<Item>{  private Logger logger = Logger.getLogger(Memcache.class.getName());  //利用LinkedHashMap实现LRU  private static LinkedHashMap<String, Item> cache;  private final int maxSize;  //负载因子  private final float DEFAULT_LOAD_FACTOR = 0.75f;  public Memcache(final int maxSize){    this.maxSize = maxSize;    //确保cache不会在达到maxSize之后自动扩容    int capacity = (int) Math.ceil(maxSize / DEFAULT_LOAD_FACTOR) + 1;    this.cache = new LinkedHashMap<String, Item>(capacity, DEFAULT_LOAD_FACTOR, true){      @Override      protected boolean removeEldestEntry(Map.Entry<String,Item> eldest) {        if (size() > maxSize){          logger.info("缓存数量已经达到上限，会删除最近最少使用的条目");        }        return size() > maxSize;      }    };    //实现同步访问    Collections.synchronizedMap(cache);  }  public synchronized boolean isFull(){    return cache.size() >= maxSize;  }  @Override  public Item get(String key) {    Item item = cache.get(key);    if (item == null){      logger.info("缓存中key:" + key + "不存在");      return null;    }else if(item!=null && item.isExpired()){ //如果缓存过期则删除并返回null      logger.info("从缓存中读取key:" + key + " value:" + item.getValue() + "已经失效");      cache.remove(key);      return null;    }    logger.info("从缓存中读取key:" + key + " value:" + item.getValue() + " 剩余有效时间" + item.remainTime());    return item;  }  @Override  public void set(String key, Item value) {    logger.info("向缓存中写入key:" + key + " value:" + value);    cache.put(key, value);  }  @Override  public Item delete(String key) {    logger.info("从缓存中删除key:" + key);    return cache.remove(key);  }  @Override  public int size(){    return cache.size();  }  @Override  public int capacity() {    return maxSize;  }  @Override  public Iterator<Map.Entry<String, Item>> iterator() {    return cache.entrySet().iterator();  }}

/** * 缓存管理器 * 后台线程 * 将cache中过期的缓存删除 */public class CacheManager implements Runnable {  private Logger logger = Logger.getLogger(CacheManager.class.getName());  //缓存  public Cache<Item> cache;  public CacheManager(Cache<Item> cache){    this.cache = cache;  }  @Override  public void run() {    while (true){      Iterator<Map.Entry<String, Item>> itemIterator = cache.iterator();      while (itemIterator.hasNext()){        Map.Entry<String, Item> entry = itemIterator.next();        Item item = entry.getValue();        if(item.isExpired()){          logger.info("key:" + entry.getKey() + " value" + item.getValue() + " 已经过期，从数据库中删除");          itemIterator.remove();        }      }      try {        //每隔5秒钟再运行该后台程序        TimeUnit.SECONDS.sleep(5);      } catch (InterruptedException e) {        e.printStackTrace();      }    }  }}

/** * 服务器 */public class IOServer implements Server {  private boolean stop;  //端口号  private final int port;  //服务器线程  private ServerSocket serverSocket;  private final Logger logger = Logger.getLogger(IOServer.class.getName());  //线程池，线程容量为maxConnection  private final ExecutorService executorService;  private final Cache<Item> cache;  public IOServer(int port, int maxConnection, Cache<Item> cache){    if (maxConnection<=0) throw new IllegalArgumentException("支持的最大连接数量必须为正整数");    this.port = port;    executorService = Executors.newFixedThreadPool(maxConnection);    this.cache = cache;  }  @Override  public void start() {    try {      serverSocket = new ServerSocket(port);      logger.info("服务器在端口"+port+"上启动");      while (true){        try {          Socket socket = serverSocket.accept();          logger.info("收到"+socket.getLocalAddress()+"的连接");          executorService.submit(new SocketHandler(socket, cache));        } catch (IOException e) {          e.printStackTrace();        }      }    } catch (IOException e) {      logger.log(Level.WARNING, "服务器即将关闭...");      e.printStackTrace();    } finally {      executorService.shutdown();      shutDown();    }  }  /**   * 服务器是否仍在运行   * @return   */  public boolean isRunning() {    return !serverSocket.isClosed();  }  /**   * 停止服务器   */  public void shutDown(){    try {      if (serverSocket!=null){        serverSocket.close();      }    } catch (IOException e) {      e.printStackTrace();    }  }}

/** * 处理各个客户端的连接 * 在获得end指令后关闭连接s */public class SocketHandler implements Runnable{  private static Logger logger = Logger.getLogger(SocketHandler.class.getName());  private final Socket socket;  private final Cache<Item> cache;  private boolean finish;  public SocketHandler(Socket s, Cache<Item> cache){    this.socket = s;    this.cache = cache;  }  @Override  public void run() {    try {      //获取socket输入流      final BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));      //获取socket输出流      final BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));      CommandFactory commandFactory = CommandFactory.getInstance(cache);      while (!finish){        final String commandLine = reader.readLine();        logger.info("ip:" + socket.getLocalAddress() + " 指令:" + commandLine);        if (commandLine == null || commandLine.trim().isEmpty()) {          continue;        }        //使用指令工厂获取指令实例        final Command command = commandFactory.getCommand(commandLine);        command.execute(reader, writer);        if (command.getType() == CommandType.END){          logger.info("请求关闭连接");          finish = true;        }      }    } catch (IOException e) {      e.printStackTrace();      logger.info("关闭来自" + socket.getLocalAddress() + "的连接");    } finally {      try {        if (socket != null){          socket.close();        }      } catch (IOException e) {        e.printStackTrace();      }    }  }}