memcached java上性能测试报告、分析与有关问题讨论

memcached java下性能测试报告、分析与问题讨论

我的项目原来使用静态HashMap来实现Key->Object的缓存,并且实现脏数据刷新.由于项目要改成集群部署.在单个jvm里运行的静态Hash结构已经无法处理脏数据问题.所以准备使用memcached做分布式缓存来解决.

从网上搜索到的资料来看 memcached能够接受较大量的请求.但其javaclient 由于大量使用同步语句、hashmap，读取流没有使用bufferedStream。造成性能并不好
为此在我的项目里我参考memcached的协议，自己实现一个客户端功能。经过测试后发现一些问题。

测试环境1：
windows xp home
迅驰T2250
1G内存
jdk1.5
server: memcached-1.2.1 nt版本
client: 自己编写
用于传输的pojo 40个属性，序列化后为750个byte
测试client与server同台机器

测试方法
填充测试时填充10w个pojo。创建100个任务，每任务负责向memcached存储1000个pojo。
读取测试时读取10000个pojo 创建100个任务，每任务读取100个pojo。
平均值均按照线程内取平均值后，各线程再取平均值。
在进行上述测试的时候 cpu占用均在90%－100%，根据上述测试结果，开2个线程时效率最高。

thread count	write avg/per obj	write totall	read avg/per obj	read total
10	2.404991 ms	29s	1.709544 ms	2s
5	0.704780 ms	18s	1.333013 ms	2s
2	0.262194 ms	15s	0.414683 ms	2s

测试环境2:
AIX5.2
IBM P650 Power4 1.5G(64bit) *4
内存8G
jdk1.4
memcached-1.2.1
libevnet1.1b
client自己编写
用于传输的pojo 40个属性，序列化后为750个byte
测试client与server同台机器

相同的测试方法。测试结果大跌眼睛 10线程时读写速度为200ms/per object 比我的笔记本慢100倍。就没继续测试

测试环境3：
windows2000 server
xeon 1.5*4
内存8G
jdk1.5
测试时发现cpu占用不高于20%

thread count	write avg/per obj	write total	read avg/per obj	read total
20	10.266615ns	71s	23.21283ns	15s
10	4.341574ns	41s	13.30084ns	16s
5	1.298717ns	25s	9.33258ns	18s
2	1.298717ns	21s	4.02503ns	23s

初步测试到这里发现的问题
1.是暂时没有达到网上宣传的1.5w个对象/s (目前测试cpu为瓶颈)
2.是aix下效率低的可怕，对aix不太熟悉。应该有些设置的参数，这点还要向大家请教。
3.超过2个线程效率就开始低。没有发挥多线程的优势，不知道大家在使用过程中有没有发现这一点，还是说我写的MemBufferedDriver有问题。

我的期望是读写速度均能稳定在0.1毫秒左右，至少 1w obj/s 这样才有可能不影响到现有系统的效率
另外大家在java下使用memcached的时候效率怎么样都是怎么用的呢

下面贴下我的client的实现
memcached的协议可以参考memcache包的doc/protocol.txt

主要是socket初次打开后不关闭，直接存放到ThreadLocal中，与memcached保持一个长练接。每次使用的时候判断连接可用还是需要重新连接。

其他存储和读取实现memcached协议。使用BufferedStream。
序列化采用实现Serializable，直接使用ObjectStream来实现。(由于都是pojo简单数据对象，尝试过实现Externalizable接口自己实现序列化和使用750个byte的String 来序列化，发现性能相差不多故放弃)

java 代码
MemBufferedDriver 为client实现

public class MemBufferedDriver {
/**
* 存放连到cacheserver的socket连接
*/
private final static ThreadLocal sockPool = new ThreadLocal();
private static String serverAddress = "localhost:11211";
public static final byte[] BYTE_GET = new byte[]{103, 101, 116, 32};
public static final byte[] BYTE_SET = new byte[]{115, 101, 116, 32};
public static final byte[] BYTE_DELETE = new byte[]{100, 101, 108, 101, 116, 101, 32};
public static final byte[] BYTE_CRLF = new byte[]{13, 10};
public static final byte[] BYTE_SPACE = new byte[]{32};
public static final String SERVER_STATUS_DELETED = "DELETED";
public static final String SERVER_STATUS_NOT_FOUND = "NOT_FOUND";
public static final String SERVER_STATUS_STORED = "STORED";
public static final String SERVER_STATUS_ERROR = "ERROR";
public static final String SERVER_STATUS_END = "END";
public static final String SERVER_STATUS_VALUE = "VALUE";
public static final String ENCODING_TYPE = "UTF-8";
public static Socket getSocket() throws UnknownHostException, IOException {
Socket s = (Socket) MemBufferedDriver.sockPool.get();
if (s == null || s.isClosed()) {
s = MemBufferedDriver.reconnect();
MemBufferedDriver.sockPool.set(s);
}
return s;
}
private static Socket reconnect() throws UnknownHostException, IOException {
String[] ip = MemBufferedDriver.serverAddress.split(":");
return new Socket(ip[0], Integer.parseInt(ip[1]));
}
public Map getMulti(String[] keys) {
Map map = new HashMap();
if (keys == null || keys.length <= 0) return map;
for (int i = 0; i < keys.length; i++) {
Object o = get(keys[i]);
if (o != null) map.put(keys[i], o);
}
return map;
}
public Object[] getMultiArray(String[] keys) {
if (keys == null || keys.length <= 0) return null;
Object[] o = new Object[keys.length];
for (int i = 0; i < keys.length; i++)
o[i] = get(keys[i]);
return o;
}
public boolean set(String key, Object obj) {
try {
if (obj == null || key == null || "".equals(key)) throw new Exception("对象和key 不能为空");
Socket s = MemBufferedDriver.getSocket();
BufferedInputStream in = new BufferedInputStream(s.getInputStream());
BufferedOutputStream out = new BufferedOutputStream(s.getOutputStream());
key = encodeKey(key);
int flag = 0;
//序列化对象
byte[] bs = object2Byte(obj);
out.write(MemBufferedDriver.BYTE_SET); //write cmd
out.write(key.getBytes()); //write key
out.write(MemBufferedDriver.BYTE_SPACE);
out.write(String.valueOf(flag).getBytes()); //write flag
out.write(MemBufferedDriver.BYTE_SPACE);
out.write("0".getBytes()); //write expire date
out.write(MemBufferedDriver.BYTE_SPACE);
out.write(String.valueOf(bs.length).getBytes()); //object length
out.write(MemBufferedDriver.BYTE_CRLF);
out.write(bs);
out.write(MemBufferedDriver.BYTE_CRLF);
out.flush();
String ret = readLine(in);
return MemBufferedDriver.SERVER_STATUS_STORED.equals(ret);
} catch (Exception e) {
System.out.println(e.getMessage());
return false;
}
}
public Object get(String key) {
try {
Socket s = MemBufferedDriver.getSocket();
InputStream in = s.getInputStream();
OutputStream out = s.getOutputStream();
key = encodeKey(key);
out.write(MemBufferedDriver.BYTE_GET);
out.write(key.getBytes());
out.write(MemBufferedDriver.BYTE_CRLF);
out.flush();
return getObjectFromStream(in, out);
} catch (Exception e) {
System.out.println(e.getMessage());
return null;
}
}
public boolean delete(String key) {
try {
Socket s = MemBufferedDriver.getSocket();
InputStream in = s.getInputStream();
OutputStream out = s.getOutputStream();
key = encodeKey(key);
out.write(MemBufferedDriver.BYTE_DELETE);
out.write(key.getBytes());
out.write(MemBufferedDriver.BYTE_CRLF);
out.flush();
String ret = readLine(in);
return MemBufferedDriver.SERVER_STATUS_DELETED.equals(ret) || MemBufferedDriver.SERVER_STATUS_NOT_FOUND.equals(ret);
} catch (Exception e) {
return false;
}
}
private Object getObjectFromStream(InputStream in, OutputStream out) throws IOException, ClassNotFoundException {
String cmd = readLine(in);
if (cmd.startsWith(MemBufferedDriver.SERVER_STATUS_VALUE)) {
//return object
String[] part = cmd.split(" ");
String para = part[2];
int length = Integer.parseInt(part[3]);
byte[] bs = new byte[length];
int count = 0;
while (count < bs.length) count += in.read(bs, count, (bs.length - count));
if (count != bs.length)
throw new IOException("读取数据长度错误");
readLine(in);
String endstr = readLine(in);
if (MemBufferedDriver.SERVER_STATUS_END.equals(endstr))
return this.byte2Object(bs);
else
System.out.println("结束标记错误");
}
return null;
}
private String encodeKey(String key) throws UnsupportedEncodingException {
return URLEncoder.encode(key, MemBufferedDriver.ENCODING_TYPE);
}
private String readLine(InputStream in) throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
boolean eol = false;
byte[] b = new byte[1];
while (in.read(b, 0, 1) != -1) {
if (b[0] == 13) eol = true;
else if (eol && b[0] == 10) break;
else
eol = false;
bos.write(b, 0, 1);
}
if (bos.size() == 0) return null;
return bos.toString().trim();
}
private byte[] object2Byte(Object o) throws IOException {
ByteArrayOutputStream b = new ByteArrayOutputStream();
new ObjectOutputStream(b).writeObject(o);
return b.toByteArray();
}
private Object byte2Object(byte[] b) throws IOException, ClassNotFoundException {
return new ObjectInputStream(new ByteArrayInputStream(b)).readObject();
}
public static void main(String[] args) throws Exception {
MemBufferedDriver m = new MemBufferedDriver();
System.out.println(m.set("a", "DsSD"));
System.out.println(m.get("a"));
}
public static void setServerAddress(String serverAddress) {
MemBufferedDriver.serverAddress = serverAddress;
}
}

java 代码
写入测试类

public class Fill2Server extends Thread {
public static int THREAD_COUNT = 2;
public static Queue queue = new Queue();
MemBufferedDriver md = new MemBufferedDriver();
public static void main(String[] args) throws Exception {
int size ;
if (args.length == 3 && args[0] != null && args[1] != null) {
MemDriver.setServerAddress(args[0]);
size = Integer.parseInt(args[1]);
THREAD_COUNT = Integer.parseInt(args[2]);
new Fill2Server().doFill(size);
} else
System.out.println("参数1 连接服务器地址 ipaddress:port ,参数2填充数量,不能小于10000,参数3为使用的线程数");
}
private void doFill(int size) throws InterruptedException {
int taskCount = size / 1000; //每个线程负责填充1000的对象
for (int i = 0; i < taskCount; i++) {
Task t = new Task();
t.setTaskId(String.valueOf(i));
queue.add(t);
}
long time = System.currentTimeMillis();
Thread tr[] = new Thread[THREAD_COUNT];
for (int i = 0; i < THREAD_COUNT; i++) {
FillThread ft = new FillThread();
(tr[i] = new Thread(ft)).start();
}
//监控填充完成
while (true) {
boolean flag = true;
for (int i = 0; i < THREAD_COUNT; i++)
flag &= tr[i].isAlive();
if (!flag) break;
Thread.sleep(1000);
}
time = System.currentTimeMillis() - time;
System.out.println("任务完成,共用" + (time / 1000) + "s");
}
class FillThread implements Runnable {
public void run() {
Task task;
while (true) {
task = (Task) queue.get();
if (task == null) break;
long time = System.nanoTime();
for (int i = 0; i < 1000; i++) {
TestBO b = new TestBO();
md.set(task.getTaskId() + i, b);
}
time = System.nanoTime() - time;
System.out.println(Thread.currentThread().getName() + " avg " + (time / 1000) + " ns ");
}
}
}
}

java 代码
读取的测试方法

public class GetFromServer extends Thread {
public static int THREAD_COUNT = 2;
public static Queue queue = new Queue();
MemDriver md = new MemDriver();
public static void main(String[] args) throws Exception {
int size;
if (args.length == 3 && args[0] != null && args[1] != null) {
MemDriver.setServerAddress(args[0]);
size = Integer.parseInt(args[1]);
THREAD_COUNT = Integer.parseInt(args[2]);
new GetFromServer().doFill(size);
} else
System.out.println("参数1 连接服务器地址 ipaddress:port ,参数2读取数量不能小于1000,参数3为使用的线程数");
}
private void doFill(int size) throws InterruptedException {
int taskCount = size / 100; //每个线程负责填充1000的对象
for (int i = 0; i < taskCount; i++) {
Task t = new Task();
t.setTaskId(String.valueOf(i));
GetFromServer.queue.add(t);
}
long time = System.currentTimeMillis();
Thread tr[] = new Thread[GetFromServer.THREAD_COUNT];
for (int i = 0; i < GetFromServer.THREAD_COUNT; i++) {
GetFromServer.FillThread ft = new GetFromServer.FillThread();
(tr[i] = new Thread(ft)).start();
}
//监控填充完成
while (true) {
boolean flag = true;
for (int i = 0; i < GetFromServer.THREAD_COUNT; i++)
flag &= tr[i].isAlive();
if (!flag) break;
Thread.sleep(1000);
}
time = System.currentTimeMillis() - time;
System.out.println("任务完成,共用" + (time / 1000) + "s");
}
class FillThread implements

13 楼 ftv_2001 2007-03-21

robbin 写道

自己用cookie id ＋MemCached的解决方案取代HttpSession

那就说在Session Scope 里面只保存了用户信息？如果是这样，那么对MemCached的利用率好像低了点。
我怎么觉得如果能够重写HttpSession或者相关的实现，第一，可以充分提高性能，第二，遗留的一些应用可以不加修改的重新部署。

14 楼 robbin 2007-03-21

ftv_2001 写道

robbin 写道

自己用cookie id ＋MemCached的解决方案取代HttpSession

我说过了，用cookie id ＋cache，根本就不用session，哪有session scope？

不同的应用服务器的HttpSession的实现都不一样，我不反对你这样去做，但是这样做一来代码丧失了应用服务器兼容性，二来某些应用服务器的HttpSession实现也不是你想像的那么好写的，你不信的话可以自己试试看。而我的方案只不过几行代码而已。

15 楼 ohmyzi 2007-03-22

codeutil 写道

昨天听同学说他的memecache用的是12G内存.存储大约1000万个键值对,查询速度很快.

浪费啊千万键值也用不了2G的内存阿难道他起了六个？

16 楼 garnoopy 2007-03-22

robbin 写道

ftv_2001 写道

robbin 写道

自己用cookie id ＋MemCached的解决方案取代HttpSession

请问使用cookie id，有没有遇到安全问题？有没有必要给cookie id加密、解密？

17 楼 youcai 2007-03-22

garnoopy 写道

请问使用cookie id，有没有遇到安全问题？有没有必要给cookie id加密、解密？

安全性等同于session,因为session使用的也是cookie id作为标识的。

18 楼 xiaoheng 2007-03-22

socket.setTcpNoDelay(true);

这样性能应该可以提高。

19 楼 ftv_2001 2007-03-23

robbin 写道

ftv_2001 写道

robbin 写道

自己用cookie id ＋MemCached的解决方案取代HttpSession

不能说你没有session Scope吧，只能说你的session不是java的session,而是更换成为了memCached内缓存，如果你有用户logout了，或者过期了，你是不是也要invalidate一下（做delete）？
我相信robbin用memcached的目的，无非就是为了解决集群时java session 复制性能低下问题。那么用memCached也就是提供一个集中式的缓存。换句话说，就是一个集中的session控制器。
当然，如果你拿来做Application 级别的cache,那就令当别论了。
另外，我觉得session这个概念的存在，是有非常的价值，你完全把session抛弃了，只能说某些应用比较合适，在某些应用下就很棘手，现在有了一个集中式的session控制器做为替代，所以我觉得怎么样无逢替代（兼容性，性能），是很有价值的。

20 楼 hiver 2007-03-23

ls,用户信息也不一定要存在session吧，完全可以map.put(cookieId, user);

21 楼 diogin 2007-03-23

session无非服务端一持久数据而已，通过sessionid这个cookie来做标识。通过memcache实现session有很多好处，除速度外，它还可以自动清除expired掉的记录。服务器集群时，本地session根本没戏，这时候完全可以把session搞到memcached里去，用sessionid当键名。如果还同时使用了cache，最好把缓存和session的数据分开成两个memcached daemon，否则cache一清扫，session数据也全没了。

22 楼 alin_ass 2007-03-27

不是很喜欢memcache java client这种搞个连接池，每次发送的时候拿到一个连接，发送后阻塞在那里等回应，一般remote cache如果每次请求带个id的话，是可以在client端用一个链接做异步的,也就是request pool而不是connection pool

23 楼 firebody 2007-03-28

robbin 写道

JavaEye现在也用到了memcache，用来做对象缓存和少量查询缓存，负载很低很低，CPU一般只有0.1%。

我使用memcache的java client来取代Java自己的HttpSession机制，然后对一个实际的web项目使用loadrunner进行压力测试，模拟20个用户，50个用户同时使用。压力测试结果表明，使用memcache取代httpsession几乎没有性能损失。(memcache java client要开socket pool)

原来使用httpSession的话，在这个压力下性能如何？

24 楼 robbin 2007-03-28

firebody 写道

robbin 写道

原来使用httpSession的话，在这个压力下性能如何？

性能几乎是一样的。

25 楼 alin_ass 2007-03-28

这个帖子里的朋友有兴趣去写一个request pool而不是connection pool的mem client么？

或者谁能证明n个链接发数据和1个链接在数据吞吐量上有什么差别, connection pool只是为了编程上方便的做法

26 楼 xieke 2007-03-30

memcache 对比其他一些缓存系统如 JCS ,OSCache，有什么优势呢？
好像都是支持集群的吧？

27 楼 grantbb 2007-04-19

xieke 写道

memcache 对比其他一些缓存系统如 JCS ,OSCache，有什么优势呢？
好像都是支持集群的吧？

上面已经说了，memcache可以作为一种集中式的cache，而OSCache可以支持集群，但是集群中服务器多，数据更新比较频繁的情况下，集群见的数据广播也是很恐怖的意见事情。

28 楼 codeutil 2007-04-20

ｑｑ新做的网站：http://city.qzone.qq.com ( 绝非作广告)

上面的搜索结果就是用的６台服务器的memcached来做的．
将搜索条件拼成key，然后去命中缓存．
每次查询的时间都在0.0x秒以下．

29 楼 zzeric 2007-04-22

用memcached做集中session处理，是否容易造成单点故障？用双机解决？

30 楼 iunknown 2007-05-17

alin_ass 写道

connection pool 的确是方便了编程，有点像服务器端的 OneThreadPerConnection 的编程风格。
如果转为 request pool ，就相当于是 event-driven 的编程风格了。这种 event-driven 风格的 api 对于使用者来说会带来比较大的困难，会增加客户端的编程难度。如果这个客户端是一个需求改变比较频繁，代码修改比较频繁的模块，那么这些增加的编程难度将带来很大的障碍。
另外对于 http 这种 request/response 方式的应用来说，在服务器端处理请求的时候，即使使用 request pool ，用异步方式来访问 memcached ，还是一样要在处理请求的 action 里等待这个异步操作的完成，相当于还是同步阻塞的，和 connection pool 没很大区别。如果客户端是一个批处理模式的应用，那么使用异步处理方式会快很多。

引用

要实现 request pool 的 memclient，需要 memcached 服务器的支持。目前的 memcached 的协议并不支持。
因为要实现 request pool 的方式，这个 RequestID 是服务器必须要知道。memcached 的协议并不支持传递 RequetID 或者产生 RequestID。
服务器必须要知道 RequestID，是因为服务器必须要在 Response 中加上这个 RequestID ，这样客户端才能把 response 和 request 对应起来。RequestID 可以由客户端生成，传递给服务器，也可以由服务器在收到请求之后，产生一个 RequestID ，客户端每次发送请求之后，同步读这个 RequestID 。通常如果请求的处理时间比较长，可以考虑由服务器来产生，否则读取服务器产生的 RequestID 的消耗可能已经超过读取本来的 Response 了。

31 楼 MiMiEye 2007-05-31

timelyRain 写道： 
<div class='quote_div'>
我的项目原来使用静态HashMap来实现Key->Object的缓存,并且实现脏数据刷新.由于项目要改成集群部署.在单个jvm里运行的静态Hash结构已经无法处理脏数据问题.所以准备使用memcached做分布式缓存来解决.
从网上搜索到的资料来看 memcached能够接受较大量的请求.但其javaclient 由于大量使用同步语句、hashmap，读取流没有使用bufferedStream。造成性能并不好 
为此在我的项目里我参考memcached的协议，自己实现一个客户端功能。经过测试后发现一些问题。
测试环境1： 
windows xp home 
迅驰T2250 
1G内存 
jdk1.5 
server: memcached-1.2.1 nt版本 
client: 自己编写 
用于传输的pojo 40个属性，序列化后为750个byte 
测试client与server同台机器
测试方法 
填充测试时填充10w个pojo。创建100个任务，每任务负责向memcached存储1000个pojo。 
读取测试时读取10000个pojo 创建100个任务，每任务读取100个pojo。 
平均值均按照线程内取平均值后，各线程再取平均值。 
在进行上述测试的时候 cpu占用均在90%－100%，根据上述测试结果，开2个线程时效率最高。

<table cellspacing='1' border='1' align='left' summary='' style='width: 520px; height: 107px;' cellpadding='1' width='520'>
 <tbody>
 <tr>
 <td>thread count</td>
 <td>write avg/per obj</td>
 <td>write totall</td>
 <td>read avg/per obj</td>
 <td>read total</td>
 </tr>
 <tr>
 <td>10</td>
 <td>2.404991 ms</td>
 <td>29s</td>
 <td>1.709544 ms</td>
 <td>2s</td>
 </tr>
 <tr>
 <td>5</td>
 <td>0.704780 ms</td>
 <td>18s</td>
 <td>1.333013 ms</td>
 <td>2s</td>
 </tr>
 <tr>
 <td>2</td>
 <td>0.262194 ms</td>
 <td>15s</td>
 <td>0.414683 ms</td>
 <td>2s</td>
 </tr>
 </tbody>
</table>
 
 
 
 
 
 


测试环境2: 
AIX5.2 
IBM P650 Power4 1.5G(64bit) *4 
内存8G 
jdk1.4 
memcached-1.2.1 
libevnet1.1b 
client自己编写 
用于传输的pojo 40个属性，序列化后为750个byte 
测试client与server同台机器 
相同的测试方法。测试结果大跌眼睛 10线程时读写速度为200ms/per object 比我的笔记本慢100倍。就没继续测试
测试环境3： 
windows2000 server 
xeon 1.5*4 
内存8G 
jdk1.5 
测试时发现cpu占用不高于20% 
<table cellspacing='1' border='1' align='left' summary='' style='width: 543px; height: 118px;' cellpadding='1' width='543'>
 <tbody>
 <tr>
 <td>thread count</td>
 <td>write avg/per obj</td>
 <td>write total</td>
 <td>read avg/per obj</td>
 <td>read total</td>
 </tr>
 <tr>
 <td>20</td>
 <td>10.266615ns</td>
 <td>71s</td>
 <td>23.21283ns</td>
 <td>15s</td>
 </tr>
 <tr>
 <td>10</td>
 <td>4.341574ns</td>
 <td>41s</td>
 <td>13.30084ns</td>
 <td>16s</td>
 </tr>
 <tr>
 <td>5</td>
 <td>1.298717ns</td>
 <td>25s</td>
 <td>9.33258ns</td>
 <td>18s</td>
 </tr>
 <tr>
 <td>2</td>
 <td>1.298717ns</td>
 <td>21s</td>
 <td>4.02503ns</td>
 <td>23s</td>
 </tr>
 </tbody>
</table>
 


 

 
 
 


 
初步测试到这里发现的问题 
1.是暂时没有达到网上宣传的1.5w个对象/s (目前测试cpu为瓶颈) 
2.是aix下效率低的可怕，对aix不太熟悉。应该有些设置的参数，这点还要向大家请教。 
3.超过2个线程效率就开始低。没有发挥多线程的优势，不知道大家在使用过程中有没有发现这一点，还是说我写的MemBufferedDriver有问题。
我的期望是读写速度均能稳定在0.1毫秒左右，至少 1w obj/s 这样才有可能不影响到现有系统的效率 
另外大家在java下使用memcached的时候效率怎么样都是怎么用的呢
下面贴下我的client的实现 
memcached的协议可以参考memcache包的doc/protocol.txt
主要是socket初次打开后不关闭，直接存放到ThreadLocal中，与memcached保持一个长练接。每次使用的时候判断连接可用还是需要重新连接。 
 
其他存储和读取实现memcached协议。使用BufferedStream。 
序列化采用实现Serializable，直接使用ObjectStream来实现。(由于都是pojo简单数据对象，尝试过实现Externalizable接口自己实现序列化和使用750个byte的String 来序列化，发现性能相差不多故放弃) 


<div class='code_title'>java 代码 
MemBufferedDriver 为client实现</div>
<div class='dp-highlighter'>
<div class='bar'/>
<ol class='dp-j'>
 <li class='alt'>public class MemBufferedDriver { </li>
 <li class=''> /** </li>
 <li class='alt'> * 存放连到cacheserver的socket连接 </li>
 <li class=''> */ </li>
 <li class='alt'> private final static ThreadLocal sockPool = new ThreadLocal(); </li>
 <li class=''> private static String serverAddress = "localhost:11211"; </li>
 <li class='alt'> </li>
 <li class=''> public static final byte[] BYTE_GET = new byte[]{103, 101, 116, 32}; </li>
 <li class='alt'> public static final byte[] BYTE_SET = new byte[]{115, 101, 116, 32}; </li>
 <li class=''> public static final byte[] BYTE_DELETE = new byte[]{100, 101, 108, 101, 116, 101, 32}; </li>
 <li class='alt'> public static final byte[] BYTE_CRLF = new byte[]{13, 10}; </li>
 <li class=''> public static final byte[] BYTE_SPACE = new byte[]{32}; </li>
 <li class='alt'> </li>
 <li class=''> public static final String SERVER_STATUS_DELETED = "DELETED"; </li>
 <li class='alt'> public static final String SERVER_STATUS_NOT_FOUND = "NOT_FOUND"; </li>
 <li class=''> public static final String SERVER_STATUS_STORED = "STORED"; </li>
 <li class='alt'> public static final String SERVER_STATUS_ERROR = "ERROR"; </li>
 <li class=''> public static final String SERVER_STATUS_END = "END"; </li>
 <li class='alt'> public static final String SERVER_STATUS_VALUE = "VALUE"; </li>
 <li class=''> </li>
 <li class='alt'> public static final String ENCODING_TYPE = "UTF-8"; </li>
 <li class=''> </li>
 <li class='alt'> </li>
 <li class=''> public static Socket getSocket() throws UnknownHostException, IOException { </li>
 <li class='alt'> Socket s = (Socket) MemBufferedDriver.sockPool.get(); </li>
 <li class=''> if (s == null || s.isClosed()) { </li>
 <li class='alt'> s = MemBufferedDriver.reconnect(); </li>
 <li class=''> MemBufferedDriver.sockPool.set(s); </li>
 <li class='alt'> } </li>
 <li class=''> return s; </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> private static Socket reconnect() throws UnknownHostException, IOException { </li>
 <li class=''> String[] ip = MemBufferedDriver.serverAddress.split(":"); </li>
 <li class='alt'> return new Socket(ip[0], Integer.parseInt(ip[1])); </li>
 <li class=''> </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> </li>
 <li class=''> public Map getMulti(String[] keys) { </li>
 <li class='alt'> Map map = new HashMap(); </li>
 <li class=''> if (keys == null || keys.length <= 0) return map; </li>
 <li class='alt'> </li>
 <li class=''> for (int i = 0; i < keys.length; i++) { </li>
 <li class='alt'> Object o = get(keys[i]); </li>
 <li class=''> if (o != null) map.put(keys[i], o); </li>
 <li class='alt'> </li>
 <li class=''> } </li>
 <li class='alt'> return map; </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> public Object[] getMultiArray(String[] keys) { </li>
 <li class='alt'> if (keys == null || keys.length <= 0) return null; </li>
 <li class=''> </li>
 <li class='alt'> Object[] o = new Object[keys.length]; </li>
 <li class=''> for (int i = 0; i < keys.length; i++) </li>
 <li class='alt'> o[i] = get(keys[i]); </li>
 <li class=''> </li>
 <li class='alt'> return o; </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> public boolean set(String key, Object obj) { </li>
 <li class='alt'> try { </li>
 <li class=''> if (obj == null || key == null || "".equals(key)) throw new Exception("对象和key 不能为空"); </li>
 <li class='alt'> Socket s = MemBufferedDriver.getSocket(); </li>
 <li class=''> BufferedInputStream in = new BufferedInputStream(s.getInputStream()); </li>
 <li class='alt'> BufferedOutputStream out = new BufferedOutputStream(s.getOutputStream()); </li>
 <li class=''> </li>
 <li class='alt'> key = encodeKey(key); </li>
 <li class=''> int flag = 0; </li>
 <li class='alt'> </li>
 <li class=''> //序列化对象 </li>
 <li class='alt'> byte[] bs = object2Byte(obj); </li>
 <li class=''> </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_SET); //write cmd </li>
 <li class=''> out.write(key.getBytes()); //write key </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_SPACE); </li>
 <li class=''> out.write(String.valueOf(flag).getBytes()); //write flag </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_SPACE); </li>
 <li class=''> out.write("0".getBytes()); //write expire date </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_SPACE); </li>
 <li class=''> out.write(String.valueOf(bs.length).getBytes()); //object length </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_CRLF); </li>
 <li class=''> </li>
 <li class='alt'> out.write(bs); </li>
 <li class=''> out.write(MemBufferedDriver.BYTE_CRLF); </li>
 <li class='alt'> out.flush(); </li>
 <li class=''> </li>
 <li class='alt'> String ret = readLine(in); </li>
 <li class=''> return MemBufferedDriver.SERVER_STATUS_STORED.equals(ret); </li>
 <li class='alt'> } catch (Exception e) { </li>
 <li class=''> System.out.println(e.getMessage()); </li>
 <li class='alt'> return false; </li>
 <li class=''> } </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> public Object get(String key) { </li>
 <li class=''> try { </li>
 <li class='alt'> Socket s = MemBufferedDriver.getSocket(); </li>
 <li class=''> InputStream in = s.getInputStream(); </li>
 <li class='alt'> OutputStream out = s.getOutputStream(); </li>
 <li class=''> key = encodeKey(key); </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_GET); </li>
 <li class=''> out.write(key.getBytes()); </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_CRLF); </li>
 <li class=''> out.flush(); </li>
 <li class='alt'> </li>
 <li class=''> return getObjectFromStream(in, out); </li>
 <li class='alt'> } catch (Exception e) { </li>
 <li class=''> System.out.println(e.getMessage()); </li>
 <li class='alt'> return null; </li>
 <li class=''> } </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> public boolean delete(String key) { </li>
 <li class=''> try { </li>
 <li class='alt'> Socket s = MemBufferedDriver.getSocket(); </li>
 <li class=''> InputStream in = s.getInputStream(); </li>
 <li class='alt'> OutputStream out = s.getOutputStream(); </li>
 <li class=''> key = encodeKey(key); </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_DELETE); </li>
 <li class=''> out.write(key.getBytes()); </li>
 <li class='alt'> out.write(MemBufferedDriver.BYTE_CRLF); </li>
 <li class=''> out.flush(); </li>
 <li class='alt'> </li>
 <li class=''> String ret = readLine(in); </li>
 <li class='alt'> return MemBufferedDriver.SERVER_STATUS_DELETED.equals(ret) || MemBufferedDriver.SERVER_STATUS_NOT_FOUND.equals(ret); </li>
 <li class=''> } catch (Exception e) { </li>
 <li class='alt'> return false; </li>
 <li class=''> } </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> private Object getObjectFromStream(InputStream in, OutputStream out) throws IOException, ClassNotFoundException { </li>
 <li class=''> String cmd = readLine(in); </li>
 <li class='alt'> if (cmd.startsWith(MemBufferedDriver.SERVER_STATUS_VALUE)) { </li>
 <li class=''> //return object </li>
 <li class='alt'> String[] part = cmd.split(" "); </li>
 <li class=''> String para = part[2]; </li>
 <li class='alt'> int length = Integer.parseInt(part[3]); </li>
 <li class=''> </li>
 <li class='alt'> byte[] bs = new byte[length]; </li>
 <li class=''> </li>
 <li class='alt'> int count = 0; </li>
 <li class=''> while (count < bs.length) count += in.read(bs, count, (bs.length - count)); </li>
 <li class='alt'> if (count != bs.length) </li>
 <li class=''> throw new IOException("读取数据长度错误"); </li>
 <li class='alt'> readLine(in); </li>
 <li class=''> String endstr = readLine(in); </li>
 <li class='alt'> if (MemBufferedDriver.SERVER_STATUS_END.equals(endstr)) </li>
 <li class=''> return this.byte2Object(bs); </li>
 <li class='alt'> else </li>
 <li class=''> System.out.println("结束标记错误"); </li>
 <li class='alt'> </li>
 <li class=''> } </li>
 <li class='alt'> return null; </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> private String encodeKey(String key) throws UnsupportedEncodingException { </li>
 <li class='alt'> return URLEncoder.encode(key, MemBufferedDriver.ENCODING_TYPE); </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> </li>
 <li class='alt'> private String readLine(InputStream in) throws IOException { </li>
 <li class=''> ByteArrayOutputStream bos = new ByteArrayOutputStream(); </li>
 <li class='alt'> boolean eol = false; </li>
 <li class=''> byte[] b = new byte[1]; </li>
 <li class='alt'> while (in.read(b, 0, 1) != -1) { </li>
 <li class=''> if (b[0] == 13) eol = true; </li>
 <li class='alt'> else if (eol && b[0] == 10) break; </li>
 <li class=''> else </li>
 <li class='alt'> eol = false; </li>
 <li class=''> </li>
 <li class='alt'> bos.write(b, 0, 1); </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> </li>
 <li class='alt'> if (bos.size() == 0) return null; </li>
 <li class=''> return bos.toString().trim(); </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> </li>
 <li class=''> private byte[] object2Byte(Object o) throws IOException { </li>
 <li class='alt'> ByteArrayOutputStream b = new ByteArrayOutputStream(); </li>
 <li class=''> new ObjectOutputStream(b).writeObject(o); </li>
 <li class='alt'> return b.toByteArray(); </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> private Object byte2Object(byte[] b) throws IOException, ClassNotFoundException { </li>
 <li class='alt'> return new ObjectInputStream(new ByteArrayInputStream(b)).readObject(); </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> </li>
 <li class='alt'> public static void main(String[] args) throws Exception { </li>
 <li class=''> MemBufferedDriver m = new MemBufferedDriver(); </li>
 <li class='alt'> System.out.println(m.set("a", "DsSD")); </li>
 <li class=''> System.out.println(m.get("a")); </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> public static void setServerAddress(String serverAddress) { </li>
 <li class=''> MemBufferedDriver.serverAddress = serverAddress; </li>
 <li class='alt'> } </li>
 <li class=''>} </li>
</ol>
</div>
 
java 代码 
写入测试类
<div class='dp-highlighter'>
<div class='bar'/>
<ol class='dp-j'>
 <li class='alt'>public class Fill2Server extends Thread { </li>
 <li class=''> public static int THREAD_COUNT = 2; </li>
 <li class='alt'> public static Queue queue = new Queue(); </li>
 <li class=''> MemBufferedDriver md = new MemBufferedDriver(); </li>
 <li class='alt'> </li>
 <li class=''> public static void main(String[] args) throws Exception { </li>
 <li class='alt'> </li>
 <li class=''> int size ; </li>
 <li class='alt'> if (args.length == 3 && args[0] != null && args[1] != null) { </li>
 <li class=''> MemDriver.setServerAddress(args[0]); </li>
 <li class='alt'> size = Integer.parseInt(args[1]); </li>
 <li class=''> THREAD_COUNT = Integer.parseInt(args[2]); </li>
 <li class='alt'> new Fill2Server().doFill(size); </li>
 <li class=''> } else </li>
 <li class='alt'> System.out.println("参数1 连接服务器地址 ipaddress:port ,参数2填充数量,不能小于10000,参数3为使用的线程数"); </li>
 <li class=''> </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> private void doFill(int size) throws InterruptedException { </li>
 <li class=''> int taskCount = size / 1000; //每个线程负责填充1000的对象 </li>
 <li class='alt'> for (int i = 0; i < taskCount; i++) { </li>
 <li class=''> Task t = new Task(); </li>
 <li class='alt'> t.setTaskId(String.valueOf(i)); </li>
 <li class=''> queue.add(t); </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> long time = System.currentTimeMillis(); </li>
 <li class=''> Thread tr[] = new Thread[THREAD_COUNT]; </li>
 <li class='alt'> for (int i = 0; i < THREAD_COUNT; i++) { </li>
 <li class=''> FillThread ft = new FillThread(); </li>
 <li class='alt'> (tr[i] = new Thread(ft)).start(); </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> //监控填充完成 </li>
 <li class='alt'> while (true) { </li>
 <li class=''> boolean flag = true; </li>
 <li class='alt'> for (int i = 0; i < THREAD_COUNT; i++) </li>
 <li class=''> flag &= tr[i].isAlive(); </li>
 <li class='alt'> </li>
 <li class=''> if (!flag) break; </li>
 <li class='alt'> </li>
 <li class=''> Thread.sleep(1000); </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> time = System.currentTimeMillis() - time; </li>
 <li class=''> System.out.println("任务完成,共用" + (time / 1000) + "s"); </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> class FillThread implements Runnable { </li>
 <li class=''> public void run() { </li>
 <li class='alt'> Task task; </li>
 <li class=''> while (true) { </li>
 <li class='alt'> task = (Task) queue.get(); </li>
 <li class=''> if (task == null) break; </li>
 <li class='alt'> long time = System.nanoTime(); </li>
 <li class=''> for (int i = 0; i < 1000; i++) { </li>
 <li class='alt'> TestBO b = new TestBO(); </li>
 <li class=''> md.set(task.getTaskId() + i, b); </li>
 <li class='alt'> } </li>
 <li class=''> time = System.nanoTime() - time; </li>
 <li class='alt'> System.out.println(Thread.currentThread().getName() + " avg " + (time / 1000) + " ns "); </li>
 <li class=''> } </li>
 <li class='alt'> } </li>
 <li class=''> } </li>
 <li class='alt'>} </li>
</ol>
</div>
 
<div class='code_title'>java 代码 
读取的测试方法</div>
<div class='dp-highlighter'>
<div class='bar'/>
<ol class='dp-j'>
 <li class='alt'>public class GetFromServer extends Thread { </li>
 <li class=''> public static int THREAD_COUNT = 2; </li>
 <li class='alt'> public static Queue queue = new Queue(); </li>
 <li class=''> MemDriver md = new MemDriver(); </li>
 <li class='alt'> </li>
 <li class=''> public static void main(String[] args) throws Exception { </li>
 <li class='alt'> </li>
 <li class=''> int size; </li>
 <li class='alt'> if (args.length == 3 && args[0] != null && args[1] != null) { </li>
 <li class=''> MemDriver.setServerAddress(args[0]); </li>
 <li class='alt'> size = Integer.parseInt(args[1]); </li>
 <li class=''> THREAD_COUNT = Integer.parseInt(args[2]); </li>
 <li class='alt'> new GetFromServer().doFill(size); </li>
 <li class=''> } else </li>
 <li class='alt'> System.out.println("参数1 连接服务器地址 ipaddress:port ,参数2读取数量不能小于1000,参数3为使用的线程数"); </li>
 <li class=''> </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> private void doFill(int size) throws InterruptedException { </li>
 <li class=''> int taskCount = size / 100; //每个线程负责填充1000的对象 </li>
 <li class='alt'> for (int i = 0; i < taskCount; i++) { </li>
 <li class=''> Task t = new Task(); </li>
 <li class='alt'> t.setTaskId(String.valueOf(i)); </li>
 <li class=''> GetFromServer.queue.add(t); </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> long time = System.currentTimeMillis(); </li>
 <li class=''> Thread tr[] = new Thread[GetFromServer.THREAD_COUNT]; </li>
 <li class='alt'> for (int i = 0; i < GetFromServer.THREAD_COUNT; i++) { </li>
 <li class=''> GetFromServer.FillThread ft = new GetFromServer.FillThread(); </li>
 <li class='alt'> (tr[i] = new Thread(ft)).start(); </li>
 <li class=''> } </li>
 <li class='alt'> </li>
 <li class=''> //监控填充完成 </li>
 <li class='alt'> while (true) { </li>
 <li class=''> boolean flag = true; </li>
 <li class='alt'> for (int i = 0; i < GetFromServer.THREAD_COUNT; i++) </li>
 <li class=''> flag &= tr[i].isAlive(); </li>
 <li class='alt'> </li>
 <li class=''> if (!flag) break; </li>
 <li class='alt'> </li>
 <li class=''> Thread.sleep(1000); </li>
 <li class='alt'> } </li>
 <li class=''> </li>
 <li class='alt'> </li>
 <li class=''> time = System.currentTimeMillis() - time; </li>
 <li class='alt'> System.out.println("任务完成,共用" + (time / 1000) + "s"</li></ol></div></div>

32 楼 davexin 2007-09-01

codeutil 写道

昨天听同学说他的memecache用的是12G内存.存储大约1000万个键值对,查询速度很快.

不太相信能用到12G内存，估计4G就差不多了

memcached java上性能测试报告、分析与有关问题讨论

相关推荐