当前位置: 开发笔记 > 数据库 > 正文

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

作者：我还没公主 | 来源：互联网 | 2022-07-23 16:03

本篇文章主要介绍了java实现memcache服务器的示例代码，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧

什么是Memcache？

Memcache集群环境下缓存解决方案

Memcache是一个高性能的分布式的内存对象缓存系统，通过在内存里维护一个统一的巨大的hash表，它能够用来存储各种格式的数据，包括图像、视频、文件以及数据库检索的结果等。简单的说就是将数据调用到内存中，然后从内存中读取，从而大大提高读取速度。　　

Memcache是danga的一个项目，最早是LiveJournal 服务的，最初为了加速 LiveJournal 访问速度而开发的，后来被很多大型的网站采用。　　

Memcached是以守护程序方式运行于一个或多个服务器中，随时会接收客户端的连接和操作

为什么会有Memcache和memcached两种名称？

其实Memcache是这个项目的名称，而memcached是它服务器端的主程序文件名，知道我的意思了吧。一个是项目名称，一个是主程序文件名，在网上看到了很多人不明白，于是混用了。

Memcached是高性能的，分布式的内存对象缓存系统，用于在动态应用中减少数据库负载，提升访问速度。Memcached由Danga Interactive开发，用于提升LiveJournal.com访问速度的。LJ每秒动态页面访问量几千次，用户700万。Memcached将数据库负载大幅度降低，更好的分配资源，更快速访问。

这篇文章将会涉及以下内容：

Java Socket多线程服务器
Java IO
Concurrency
Memcache特性和协议

Memcache

Memcache is an in-memory key-value store for small chunks of arbitrary data (strings, objects) from results of databasecalls, API calls, or page rendering.

即内存缓存数据库，是一个键值对数据库。该数据库的存在是为了将从其他服务中获取的数据暂存在内存中，在重复访问时可以直接从命中的缓存中返回。既加快了访问速率，也减少了其他服务的负载。这里将实现一个单服务器版本的Memcache，并且支持多个客户端的同时连接。

客户端将与服务器建立telnet连接，然后按照Memcache协议与服务器缓存进行交互。这里实现的指令为get，set和del。先来看一下各个指令的格式

set

set属于存储指令，存储指令的特点时，第一行输入基本信息，第二行输入其对应的value值。

set [noreply]\r\n
\r\n

如果存储成功，将会返回STORED，如果指令中包含noreply属性，则服务器将不会返回信息。

该指令中每个域的内容如下：

key: 键
flags: 16位无符号整数，会在get时随键值对返回
exptime: 过期时间，以秒为单位
bytes：即将发送的value的长度
noreply：是否需要服务器响应，为可选属性

如果指令不符合标准，服务器将会返回ERROR。

get

get属于获取指令，该指令特点如下：

get *\r\n

它支持传入多个key的值，如果缓存命中了一个或者多个key，则会返回相应的数据，并以END作为结尾。如果没有命中，则返回的消息中不包含该key对应的值。格式如下：

VALUE   \r\n
\r\n
VALUE   \r\n
\r\n
END
del

删除指令，该指令格式如下：

del  [noreply]\r\n

如果删除成功，则返回DELETED\r\n，否则返回NOT_FOUND。如果有noreply参数，则服务器不会返回响应。

JAVA SOCKET

JAVA SOCKET需要了解的只是包括TCP协议，套接字，以及IO流。这里就不详细赘述，可以参考我的这系列文章，也建议去阅读JAVA Network Programming。一书。

代码实现

这里贴图功能出了点问题，可以去文末我的项目地址查看类图。

这里采用了指令模式和工厂模式实现指令的创建和执行的解耦。指令工厂将会接收commandLine并且返回一个Command实例。每一个Command都拥有execute方法用来执行各自独特的操作。这里只贴上del指令的特殊实现。

 /**
 * 各种指令
 * 目前支持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 memcache;
  private CommandFactory(){}

  public static CommandFactory getInstance(Cache 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 cache;

  private String key;
  private boolean noReply;
  public DeleteCommand(final String command, final Cache 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;
  }
}

然后是实现内存服务器，为了支持先进先出功能，这里使用了LinkedTreeMap作为底层实现，并且重写了removeOldest方法。同时还使用CacheManager的后台线程及时清除过期的缓存条目。

public class Memcache implements Cache{
  private Logger logger = Logger.getLogger(Memcache.class.getName());
  //利用LinkedHashMap实现LRU
  private static LinkedHashMap 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(capacity, DEFAULT_LOAD_FACTOR, true){
      @Override
      protected boolean removeEldestEntry(Map.Entry 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> iterator() {
    return cache.entrySet().iterator();
  }
}

/**
 * 缓存管理器
 * 后台线程
 * 将cache中过期的缓存删除
 */
public class CacheManager implements Runnable {

  private Logger logger = Logger.getLogger(CacheManager.class.getName());

  //缓存
  public Cache cache;

  public CacheManager(Cache cache){
    this.cache = cache;
  }


  @Override
  public void run() {
    while (true){
      Iterator> itemIterator = cache.iterator();
      while (itemIterator.hasNext()){
        Map.Entry 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();
      }

    }
  }
}

最后是实现一个多线程的Socket服务器，这里就是将ServerSocket绑定到一个接口，并且将accept到的Socket交给额外的线程处理。

/**
 * 服务器
 */
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 cache;
  public IOServer(int port, int maxConnection, Cache 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 cache;

  private boolean finish;


  public SocketHandler(Socket s, Cache 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();
      }
    }
  }
}

项目地址请戳这里，如果觉得还不错的话，希望能给个星哈><

参考资料

memcached官网
memcache协议

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。

推荐阅读

memcached
Centos下安装memcached+memcached教程

本文介绍了在Centos下安装memcached和使用memcached的教程，详细解释了memcached的工作原理，包括缓存数据和对象、减少数据库读取次数、提高网站速度等。同时，还对memcached的快速和高效率进行了解释，与传统的文件型数据库相比，memcached作为一个内存型数据库，具有更高的读取速度。 ... [详细]

蜡笔小新 2023-12-10 17:10:24
memcached
开发笔记:Memcached高性能内存对象缓存系统

篇首语：本文由编程笔记#小编为大家整理，主要介绍了Memcached高性能内存对象缓存系统相关的知识，希望对你有一定的参考价值。一、Memcached概述 ... [详细]

蜡笔小新 2023-10-13 19:08:11
memcached
DEDECMS最新5.7版在Windows下的Memcache安装

一，织梦后台后台设置进入系统后台,在[系统基本参数]下面的性能选项卡当中,关于memcache进行如下配置:cfg_memcache_enable:是否启用memcache缓存，如果为否(N) ... [详细]

蜡笔小新 2023-10-13 10:45:42
memcached
内存交换机制

相对于内存来说,磁盘的容量是非常大的,所以Linux内核实现了一个叫内存交换的功能--把某些进程的一些暂时用不到的内存页保存到磁盘中,然后把物理内存页分配给更紧急的用户使用,当 ... [详细]

蜡笔小新 2023-10-12 19:20:04
memcached
Netty拆包粘包问题解决 —— 特殊结束符

本文介绍了解决Netty拆包粘包问题的一种方法——使用特殊结束符。在通讯过程中，客户端和服务器协商定义一个特殊的分隔符号，只要没有发送分隔符号，就代表一条数据没有结束。文章还提供了服务端的示例代码。 ... [详细]

蜡笔小新 2023-12-14 18:02:45
memcached
如何找到并终止在8080端口上运行的进程？

本文介绍了如何找到并终止在8080端口上运行的进程的方法，通过使用终端命令lsof -i :8080可以获取在该端口上运行的所有进程的输出，并使用kill命令终止指定进程的运行。 ... [详细]

蜡笔小新 2023-12-14 13:45:13
memcached
Harmony 与 Game Space 达成合作，在 Shard1 上扩展 Web3 游戏

旧金山20 ... [详细]

蜡笔小新 2023-10-17 19:51:46
sql
像跟踪分布式服务调用那样跟踪Go函数调用链 | Gopher Daily (2020.12.07) ʕ◔ϖ◔ʔ

每日一谚：“Acacheisjustamemoryleakyouhaven’tmetyet.”—Mr.RogersGo技术专栏“改善Go语⾔编程质量的50个有效实践” ... [详细]

蜡笔小新 2023-10-17 19:23:45
memcached
实战项目memcached+tomcat+session+nginx在工作中的应用和配置

环境介绍：公司根据实际需要搭建一个购物网站，当用户购物时可以将不同商品，放到同一个购物车中进行同时付款。环境的搭建：外网用户IP地址：1.1.1.1主机名:fanxiaohui用户 ... [详细]

蜡笔小新 2023-10-13 07:29:54
memcached
2015第44周六tomcat集群了解

对于WEB应用集群的技术实现而言，最大的难点就是如何能在集群中的多个节点之间保持数据的一致性，会话（Session）信息是这 ... [详细]

蜡笔小新 2023-10-13 05:50:24
memcached
nagios插件内存监控

为什么80%的码农都做不了架构师？#!usrbinenvpython#-*-coding:utf-8-*-NagiosplugintoreportMe ... [详细]

蜡笔小新 2023-10-12 18:15:59
memcached
基于内存和 Redis 的两级 Java 缓存框架！

👇👇关注后回复“进群”，拉你进程序员交流群👇👇Java两级缓存框架，可以让应用支持两级缓 ... [详细]

蜡笔小新 2023-10-12 17:09:04
memcached
Memcache Session Manager

转自：MSMIntroduction如果为了简单使用，你只需要安装一个tomcat（6或者7）和memcached，在生产环境中可能会有多台tomcat服务器以及多台可用的memc ... [详细]

蜡笔小新 2023-10-12 13:00:04
memcached
开发笔记:ansible角色调用

本篇博客仅供学习，没有实际项目意义，仅供学习之用实验目的:配置中心服务器自动配置两台机器分别部署为n ... [详细]

蜡笔小新 2023-10-12 12:21:57
memcached
php套件（php套件安装）

本文目录一览：1、常用的php环境套件有哪些2 ... [详细]

蜡笔小新 2023-10-12 11:59:50

我还没公主

这个家伙很懒，什么也没留下！

Tags | 热门标签

RankList | 热门文章