为了分析在插入海量数据到Cassandra集群或者Oracle时的表现，也就是插入速率，我们用java程序对插入数据的用时进行了采样，最终用JFreeChart把采样结果绘制出来了。

为了公平起见，我们做了以下处理：

1.所有的循环变量都放在了循环外面

2.对于Cassandra的replication-factor设置为1，这样插入数据不需要插入额外的备份。

3.对于Oracle我们用预编译语句，这样插入操作的执行计划可以重用。

4.所有的测试都在周末进行，这样不可能有其他人去干扰这些服务器。

5.这些机器上运行的其他进程都被我kill掉了，这样保证CPU，内存的专用性。

6.在用java代码插入Cassandra记录时候，我采用了thrift API, 因为它的效率比Hector API高。

以下是实验（分两部分，一是采样部分，二是数据分析部分）

Part 1:采样：

Cassandra的采样：

我们这里依然用循环插入50W条记录，不同的是，在循环的开始和循环每10000条记录时，我们把时间戳记录在List中，最终把这个List写入文本文件(cassandra_input_sample_data.txt)：

1. package com.charles.cassandra.demo;
2. import java.io.File;
3. import java.io.FileWriter;
4. import java.util.ArrayList;
5. import java.util.List;
6. import org.apache.cassandra.thrift.Cassandra;
7. import org.apache.cassandra.thrift.Column;
8. import org.apache.cassandra.thrift.ColumnParent;
9. import org.apache.cassandra.thrift.ConsistencyLevel;
10. import org.apache.cassandra.thrift.TBinaryProtocol;
11. import org.apache.thrift.protocol.TProtocol;
12. import org.apache.thrift.transport.TFramedTransport;
13. import org.apache.thrift.transport.TSocket;
14. import org.apache.thrift.transport.TTransport;
15. import com.charles.cassandra.util.CassandraOperationUtil;
16. public class CassandraClusterStressTest
17. {
18. public static void main(String[] args)
19. throws Exception
20. {
21. //包装好的socket
22. TTransport tr = new TFramedTransport(new TSocket("192.168.129.34",9160));
23. TProtocol proto = new TBinaryProtocol(tr);
24. Cassandra.Client client = new Cassandra.Client(proto);
25. tr.open();
26. if(!tr.isOpen())
27. {
28. System.out.println("无法连接到服务器!");
29. return;
30. }
31. System.out.println("开始压力测试，我们插入50W条数据到2节点集群中");
32. System.out.println("...");
33. //标记开始时间
34. long startTime = System.currentTimeMillis();
35. client.set_keyspace("Charles_Stress_Test2");//使用Charles_Stress_Test keyspace
36. ColumnParent parent = new ColumnParent("student");//column family
37. /*
38. * 这里我们插入50万条数据到Student内
39. * 每条数据包括id和name
40. */
41. String key_user_id = "a";
42. String k;
43. long timestamp;
44. Column idColumn =null;
45. Column nameColumn=null;
46. //这个sampleData代表了每插入1W条记录到Cassandra集群的用时毫秒的数据样本
47. List<Integer> sampleData = new ArrayList<Integer>(51);
48. for(int i = 0;i < 500000;i++)
49. {
50. k = key_user_id + i;//row key
51. timestamp = System.currentTimeMillis();//时间戳
52. //每行的第一个字段（id字段）
53. idColumn = new Column(CassandraOperationUtil.stringToByteBuffer("id"));//字段名
54. idColumn.setValue(CassandraOperationUtil.stringToByteBuffer(i + ""));//字段值
55. idColumn.setTimestamp(timestamp);//时间戳
56. client.insert(
57. CassandraOperationUtil.stringToByteBuffer(k),
58. parent,
59. idColumn,
60. ConsistencyLevel.ONE);
61. //每行的第二个字段(name字段）
62. nameColumn = new Column(CassandraOperationUtil.stringToByteBuffer("name"));
63. nameColumn.setValue(CassandraOperationUtil.stringToByteBuffer("student" + i));
64. nameColumn.setTimestamp(timestamp);
65. client.insert(
66. CassandraOperationUtil.stringToByteBuffer(k),
67. parent,
68. nameColumn,
69. ConsistencyLevel.ONE);
70. //判断是否这是起始记录（用于标记起始时间戳）和第N 万条记录（第N万条记录的时间戳）
71. if( (i==0) || ( (i+1)%10000==0)){
72. sampleData.add((int)(timestamp));
73. }
74. }
75. //标记结束时间
76. long endTime = System.currentTimeMillis();
77. //标记一共用时
78. long elapsedTime = endTime-startTime;
79. System.out.println("压力测试完毕,用时: "+elapsedTime+" 毫秒");
80. //关闭连接
81. tr.close();
82. //压力测试结束后，我们把所有的样本数据写入文件中等待处理
83. FileWriter fw = new FileWriter(new File("cassandra_insert_sample_data.txt"));
84. for(int j=0;j<sampleData.size();j++){
85. fw.write(sampleData.get(j)+"\n");
86. }
87. fw.flush();
88. fw.close();
89. }
90. }

最终50W条记录插入完毕：控制台显示：

而且我们打开文本文件确定这些时间戳的样本都被记录了：

当然了，因为Cassandra的存储是基于内存的，所以我们定义了一个工具类用于转换字符串和字节数组：

1. /*
2. */
3. package com.charles.cassandra.util;
4. import java.io.UnsupportedEncodingException;
5. import java.nio.ByteBuffer;
6. /**
7. *
8. * Description: 这个类提供了一些Cassandra操作的工具类
9. *
10. * @author charles.wang
11. * @created May 19, 2012 11:18:27 AM
12. *
13. */
14. public class CassandraOperationUtil {
15. /**
16. *因为在Cassandra中，信息都存在内存的，所以都是以ByteBuffer形式存储的，但是ByteBuffer对于人类来说没有String可读性强
17. *所以这个方法可以吧字符串转为ByteBuffer
18. */
19. public static ByteBuffer stringToByteBuffer(String s) throws UnsupportedEncodingException{
20. return ByteBuffer.wrap(s.getBytes("UTF-8"));
21. }
22. /**
23. *因为在Cassandra中，信息都存在内存的，所以都是以ByteBuffer形式存储的，但是ByteBuffer对于人类来说没有String可读性强
24. *所以对称的，这个方法吧ByteBuffer转为人类可读的字符串
25. */
26. public static String byteBufferToString (ByteBuffer b) throws UnsupportedEncodingException{
27. //先构建一个字节数组
28. byte[] bytes = new byte[b.remaining()];
29. //吧bytebuffer里面的内容全部存入字节数组
30. b.get(bytes);
31. //然后把这些bytes转为String
32. return new String(bytes,"UTF-8");
33. }
34. }

Oracle的采样：

我们这里依然用循环插入50W条记录，不同的是，在循环的开始和循环每10000条记录时，我们把时间戳记录在List中，最终把这个List写入文本文件(oracle_input_sample_data.txt)：

1. /*
2. */
3. package com.charles.cassandra.demo;
4. import java.io.File;
5. import java.io.FileWriter;
6. import java.sql.Connection;
7. import java.sql.Date;
8. import java.sql.DriverManager;
9. import java.sql.PreparedStatement;
10. import java.sql.ResultSet;
11. import java.sql.Statement;
12. import java.sql.ResultSetMetaData;
13. import java.sql.Timestamp;
14. import java.util.ArrayList;
15. import java.util.List;
16. /**
17. *
18. * Description:插入50W条记录到关系数据库Oracle中
19. *
20. * @author charles.wang
21. * @created May 19, 2012 5:25:36 PM
22. *
23. */
24. public class OracleStressTest {
25. /**
26. * 既然要测负载，就尽可能减少方法调用的时间开销，所以我用了最原始的写法
27. * @param args
28. */
29. public static void main(String[] args){
30. String url="jdbc:oracle:thin:@192.168.129.14:15210:ora11g";
31. String username="Charles_Stress_Test1";
32. String password="Charles_Stress_Test1";
33. String sDBDriver = "oracle.jdbc.driver.OracleDriver";
34. try{
35. System.out.println("开始压力测试，我们以预编译的方式插入50W条数据到Oracle中");
36. System.out.println("...");
37. //标记开始时间
38. long startTime=System.currentTimeMillis();
39. Class.forName(sDBDriver).newInstance();
40. Connection conn = DriverManager.getConnection(url,username,password);
41. //因为这里使用预编译语句，所以不用每次都生成新的执行计划
42. String rowkey=null;
43. String id=null;
44. String name=null;
45. Date date=null;
46. String statementString="insert into Student (rowkey,id,name,create_date )values(?,?,?,?)";;
47. PreparedStatement pstmt = conn.prepareStatement(statementString);
48. //这个sampleData代表了每插入1W条记录到Oracle数据库用时毫秒的数据样本
49. List<Integer> sampleData = new ArrayList<Integer>(51);
50. for(int i=0;i<500000;i++){
51. long timestamp = System.currentTimeMillis();
52. rowkey="a"+i;
53. id=""+i;
54. name="student"+i;
55. date= new Date(timestamp);
56. pstmt.setString(1,rowkey);
57. pstmt.setString(2, id);
58. pstmt.setString(3,name);
59. pstmt.setDate(4, date);
60. pstmt.execute();
61. //判断是否这是起始记录（用于标记起始时间戳）和第N 万条记录（第N万条记录的时间戳）
62. if( (i==0) || ( (i+1)%10000==0)){
63. sampleData.add((int)(timestamp));
64. }
65. }
66. //关闭相关连接
67. pstmt.close();
68. conn.close();
69. long endTime=System.currentTimeMillis();
70. long elapsedTime=endTime-startTime;
71. System.out.println("压力测试完毕,用时: "+elapsedTime+" 毫秒");
72. //在压力测试结束之后，我们来把样本数据写入文本文件中
73. FileWriter fw = new FileWriter(new File("oracle_insert_sample_data.txt"));
74. for(int j=0;j<sampleData.size();j++){
75. fw.write(sampleData.get(j)+"\n");
76. }
77. fw.flush();
78. fw.close();
79. }catch(Exception e){
80. System.out.println("数据库连接失败");
81. e.printStackTrace();
82. }
83. }
84. }

最终50W条记录插入完毕：控制台显示：

而且我们打开文本文件确定这些时间戳的样本都被记录了：

Part 2： 分析采样数据并且绘制比较图：

我们用JFreechart强大的图表制作能力来绘制比较图：

首先我们依然定义一个工具类 ParseDataUtil，它可以完成两件事情，一是从样本文件中读取数据，然后时间戳相减，最终把所有每1W条数据的耗时时间存入List<Integer>对象，二是它可以吧List<Integer>对象传递给JFreechart的数据模型：

1. /*
2. */
3. package com.charles.parsedata.util;
4. import java.io.BufferedReader;
5. import java.io.File;
6. import java.io.FileInputStream;
7. import java.io.FileNotFoundException;
8. import java.io.FileReader;
9. import java.io.IOException;
10. import java.io.InputStreamReader;
11. import java.util.ArrayList;
12. import java.util.List;
13. import org.jfree.data.category.DefaultCategoryDataset;
14. /**
15. *
16. * Description:
17. *
18. * @author charles.wang
19. * @created May 21, 2012 8:45:28 AM
20. *
21. */
22. public class ParseDataUtil {
23. /**
24. * 这个方法用于添加指定的分析来的数据作为JFreechart显示的数据集
25. *
26. * @param ds
27. *            JFreechart的数据集对象
28. * @param datas
29. *            从压力测试采样并且经过加工后的数据
30. * @param seriesName
31. *            曲线的名称
32. */
33. public static void addDataToDataset(DefaultCategoryDataset ds, List<Integer> datas, String seriesName) {
34. // 对于数据集合的检查
35. if (datas.size() <= 0)
36. return;
37. // type表示横轴的每个坐标点
38. Integer value = 0;
39. String type = null;
40. // 用循环依次添加
41. for (int i = 1; i <= datas.size(); i++) {
42. // 获取每个样本数据中的横坐标纵坐标
43. type = i + "";
44. value = datas.get(i - 1);
45. ds.addValue(value, seriesName, type);
46. }
47. }
48. /**
49. * 这个方法用于从样本数据中构建最终传入到JFreechart绘制的数据
50. *
51. * @param fileName
52. * @param numOfRecords
53. * @return
54. */
55. public static List<Integer> buildSampleDataListFromFile(String fileName, int numOfRecords) {
56. // 判断参数
57. if (numOfRecords <= 0)
58. return null;
59. try {
60. // 创建一个rawSampleData 作为采样数据的List
61. List<Integer> rawSampleData = new ArrayList<Integer>(numOfRecords);
62. // 打开一个到指定文件的输入流
63. FileInputStream fis = new FileInputStream(fileName);
64. InputStreamReader isr = new InputStreamReader(fis);
65. BufferedReader br = new BufferedReader(isr);
66. if (br == null) {
67. System.out.println("样本文件不存在！");
68. return null;
69. }
70. // 依次读入
71. for (int i = 0; i < numOfRecords; i++) {
72. String rawRecord = br.readLine();
73. rawSampleData.add(Integer.parseInt(rawRecord));
74. }
75. // 读完了关闭输入流
76. br.close();
77. isr.close();
78. fis.close();
79. // 现在我们把rawSampleData转为真正可以被JFreeChart显示的SampleData
80. // 这里SampleData的每个数据都是用时，所以是当前时间戳-第一条记录的时间戳
81. int sampleDataSize = rawSampleData.size() - 1;
82. List<Integer> sampleData = new ArrayList<Integer>(sampleDataSize);
83. // 设置起始时间戳，以后每一个时间戳都要减去这个起始时间戳
84. Integer baseTimeStamp = rawSampleData.get(0);
85. // System.out.println("baseTimeStamp: "+baseTimeStamp);
86. // System.out.println("sampleDataSize: "+sampleData.size());
87. // System.out.println("hello");
88. for (int j = 0; j < sampleDataSize; j++) {
89. int time = rawSampleData.get(j + 1) - baseTimeStamp;
90. System.out.println(time);
91. sampleData.add(time);
92. }
93. return sampleData;
94. } catch (Exception ex) {
95. ex.printStackTrace();
96. return null;
97. }
98. }
99. }

然后我们有个最终执行画图的类，这个类吧从原始数据分析后的数据显示在图表上，并且作为对比，吧Oracle和Cassandra集群的数据显示在同一张表上：

1. /*
2. */
3. package com.charles.parsedata;
4. import java.util.ArrayList;
5. import java.util.List;
6. import javax.swing.JPanel;
7. import org.jfree.chart.ChartFactory;
8. import org.jfree.chart.ChartPanel;
9. import org.jfree.chart.JFreeChart;
10. import org.jfree.chart.axis.NumberAxis;
11. import org.jfree.chart.plot.CategoryPlot;
12. import org.jfree.chart.plot.PlotOrientation;
13. import org.jfree.data.category.DefaultCategoryDataset;
14. import org.jfree.ui.ApplicationFrame;
15. import org.jfree.ui.RefineryUtilities;
16. import com.charles.parsedata.util.ParseDataUtil;
17. /**
18. *
19. * Description: 用JFreechart来分析插入数据
20. *
21. * @author charles.wang
22. * @created May 21, 2012 8:38:27 AM
23. *
24. */
25. public class InsertDataStressTestDataParser extends ApplicationFrame{
26. public InsertDataStressTestDataParser(String s) {
27. super(s);
28. setContentPane(createDemoLine());
29. }
30. public static void main(String[] args)  {
31. InsertDataStressTestDataParser fjc = new InsertDataStressTestDataParser("Cassandra&Oracle插入数据对比图");
32. fjc.pack();
33. RefineryUtilities.centerFrameOnScreen(fjc);
34. fjc.setVisible(true);
35. }
36. // 生成显示图表的面板
37. public static JPanel createDemoLine(){
38. JFreeChart jfreechart = createChart(createDataset());
39. return new ChartPanel(jfreechart);
40. }
41. // 生成图表主对象JFreeChart
42. public static JFreeChart createChart(DefaultCategoryDataset linedataset) {
43. //定义图表对象
44. JFreeChart chart = ChartFactory.createLineChart("Cassandra和Oracle插入数据对比图", // chart title
45. "记录数(单位：万条）", // 横轴标签
46. "用时(毫秒)", // 纵轴标签
47. linedataset, // 传入的数据集
48. PlotOrientation.VERTICAL, // 方向
49. true, // bool变量表示是否要加入图例(legend)
50. true, // 工具集
51. false // 是否添加url
52. );
53. CategoryPlot plot = chart.getCategoryPlot();
54. // 范围轴线
55. NumberAxis rangeAxis = (NumberAxis) plot.getRangeAxis();
56. rangeAxis.setStandardTickUnits(NumberAxis.createIntegerTickUnits());
57. rangeAxis.setAutoRangeIncludesZero(true);
58. rangeAxis.setUpperMargin(0.20);
59. rangeAxis.setLabelAngle(Math.PI / 2.0);
60. return chart;
61. }
62. //生成数据
63. public static DefaultCategoryDataset createDataset() {
64. DefaultCategoryDataset ds = new DefaultCategoryDataset();
65. List<Integer> data1 = ParseDataUtil.buildSampleDataListFromFile("cassandra_insert_sample_data.txt",51);
66. List<Integer> data2 = ParseDataUtil.buildSampleDataListFromFile("oracle_insert_sample_data.txt",51);
67. ParseDataUtil.addDataToDataset(ds, data1, "Cassandra插入数据所用时间分布图");
68. ParseDataUtil.addDataToDataset(ds, data2, "Oracle插入数据所用时间分布图");
69. return ds;
70. }
71. }

最终对比图如下：

结论：

所以我们这里很清楚的看到：

(1) 无论是Cassandra集群还是Oracle，其插入操作用时都是线性的，也就是它的平均插入速率基本是恒速。

(2) 在低配置服务器上，Cassandra集群的插入数据操作耗时要高于Oracle关系数据库。

本文转自 charles_wang888 51CTO博客，原文链接：http://blog.51cto.com/supercharles888/870876，如需转载请自行联系原作者