一、什么是事务

•  原子性。即不可分割性，事务要么全部被执行，要么就全部不被执行。如果事务的所有子事务全部提交成功，则所有的数据库操作被提交，数据库状态发生转换；如果有子事务失败，则其他子事务的数据库操作被回滚，即数据库回到事务执行前的状态，不会发生状态转换。

•  一致性或可串性。事务的执行使得数据库从一种正确状态转换成另一种正确状态。

• 隔离性。在事务正确提交之前，不允许把该事务对数据的任何改变提供给任何其他事务，即在事务正确提交之前，它可能的结果不应显示给任何其他事务。

• 持久性。事务正确提交后，其结果将永久保存在数据库中，即使在事务提交后有了其他故障，事务的处理结果也会得到保存。 　　

　　运行嵌入式SQL应用程序或脚本，在可执行SQL语句第一次执行时(在建立与数据库的连接之后或在现有事务终止之后)，事务就会自动启动。在启动事务之后，必须由启动事务的用户或应用程序显式地终止它，除非使用了称为自动提交(automatic commit)的过程(在这种情况下，发出的每个单独的SQL语句被看做单个事务，它一执行就被隐式地提交了)。

　　在大多数情况下，通过执行COMMIT或ROLLBACK语句来终止事务。当执行COMMIT语句时，自从事务启动以来对数据库所做的一切更改就成为永久性的了-- 即它们被写到磁盘。当执行ROLLBACK语句时，自从事务启动以来对数据库所做的一切更改都被撤销，并且数据库返回到事务开始之前所处的状态。不管是哪种情况，数据库在事务完成时都保证能回到一致状态。

　　一定要注意一点：虽然事务通过确保对数据的更改仅在事务被成功提交之后才成为永久性的，从而提供了一般的数据库一致性，但还是须要用户或应用程序来确保每个事务中执行的SQL操作序列始终会导致一致的数据库。

二、数据库系统支持两种事务模式：

• 自动提交模式：每个SQL语句都是一个独立的事务，当数据库系统执行完一个SQL语句后，会自动提交事务。

• 手动提交模式：必须由数据库客户程序显示指定事务开始边界和结束边界。

　　注：MySQL中数据库表分为3种类型：INNODB、BDB和MyISAM，其中MyISAM不支持数据库事务。MySQL中create table 语句默认为MyISAM类型。

三、对于同时运行的多个事务，当这些事务访问数据库中相同的数据时，如果没有采取必要的隔离机制，就会导致各种并发问题，这些并发问题可归纳为以下几类：

• 第一类丢失更新：撤销一个事务时，把其他事务已提交的更新数据覆盖。　
• 脏读：一个事务读到另一个事务为提交的更新数据。
• 虚读：一个事务读到另一个事务已提交的新插入的数据。
• 不可重复读：一个事务读到另一个事务已提交的更新数据。
• 第二类丢失更新：这是不可重复读中的特例，一个事务覆盖另一个事务已提交的更新数据。　　

四、隔离级别

　　当数据库系统采用read Commited隔离级别时，会导致不可重复读喝第二类丢失更新的并发问题，可以在应用程序中采用悲观锁或乐观锁来避免这类问题。从应用程序的角度，锁可以分为以下几类：

• Serializable(串行化)：一个事务在执行过程中完全看不到其他事务对数据库所做的更新。
• Repeatable Read(可重复读)：一个事务在执行过程中可以看到其他事务已经提交的新插入的记录，但是不能看到其他事务对已有记录的更新。
• Read Commited（读已提交数据）：一个事务在执行过程中可以看到其他事务已经提交的新插入的记录，而且能看到其他事务已经提交的对已有记录的更新
• Read Uncomitted（读未提交数据）：一个事务在执行过程中可以拷打其他事务没有提交的新插入的记录，而且能看到其他事务没有提交的对已有记录的更新。

    隔离级别越高，越能保证数据的完整性和一致性，但是对并发性能的影响也越大。对于多数应用程序，可以有优先考虑把数据库系统的隔离级别设为Read Commited，它能够避免脏读，而且具有较好的并发性能。尽管它会导致不可重复读、虚读和第二类丢失更新这些并发问题，在可能出现这类问题的个别场合，可以由应用程序采用悲观锁或乐观锁来控制。

　　当数据库系统采用read Commited隔离级别时，会导致不可重复读喝第二类丢失更新的并发问题，可以在应用程序中采用悲观锁或乐观锁来避免这类问题。从应用程序的角度，锁可以分为以下几类：

　　A.悲观锁：指在应用程序中显示的为数据资源加锁。尽管能防止丢失更新和不可重复读这类并发问题，但是它会影响并发性能，因此应该谨慎地使用。 

　　B.乐观锁：乐观锁假定当前事务操作数据资源时，不回有其他事务同时访问该数据资源，因此完全依靠数据库的隔离级别来自动管理锁的工作。应用程序采用版本控制手段来避免可能出现的并发问题。

五、悲观锁有两种实现方式。

　　A.在应用程序中显示指定采用数据库系统的独占所来锁定数据资源。SQL语句：select ... for update，在Hibernate中使用get，load时如session.get(Account.class,new Long(1),LockMode,UPGRADE) 

　　B.在数据库表中增加一个表明记录状态的LOCK字段，当它取值为“Y”时，表示该记录已经被某个事务锁定，如果为“N”，表明该记录处于空闲状态，事务可以访问它。增加锁标记字段就可以实现。

　　利用Hibernate的版本控制来实现乐观锁

　　乐观锁是由程序提供的一种机制，这种机制既能保证多个事务并发访问数据，又能防止第二类丢失更新问题。

　　在应用程序中可以利用Hibernate提供的版本控制功能来视线乐观锁，OR映射文件中的<version>元素和<timestamp>都具有版本控制的功能，一般推荐采用<version>

1.  背景

在做性能测试的时候，很多人都用并发用户数来衡量系统的性能，觉得系统能支撑的并发用户数越多，系统的性能就越好；对TPS不是非常理解，也根本不知道它们之间的关系，因此非常有必要进行解释。

2.  术语定义

Ø  并发用户数：指的是现实系统中操作业务的用户，在性能测试工具中，一般称为虚拟用户数(Virutal User)，注意并发用户数跟注册用户数、在线用户数有很大差别的，并发用户数一定会对服务器产生压力的，而在线用户数只是 ”挂” 在系统上，对服务器不产生压力，注册用户数一般指的是数据库中存在的用户数。

Ø  TPS：Transaction Per Second, 每秒事务数, 是衡量系统性能的一个非常重要的指标，

3.  Vu和TPS换算

Ø  简单例子：在术语中解释了TPS是每秒事务数，但是事务时要靠虚拟用户做出来的，假如1个虚拟用户在1秒内完成1笔事务，那么TPS明显就是1；如果某笔业务响应时间是1ms,那么1个用户在1秒内能完成1000笔事务，TPS就是1000了；如果某笔业务响应时间是1s,那么1个用户在1秒内只能完成1笔事务，要想达到1000TPS，至少需要1000个用户；因此可以说1个用户可以产生1000TPS，1000个用户也可以产生1000TPS，无非是看响应时间快慢。

Ø  复杂公式：

试想一下复杂场景，多个脚本，每个脚本里面定义了多个事务（例如一个脚本里面有100个请求，我们把这100个连续请求叫做Action,只有第10个请求，第20个请求分别定义了事务10和事务20）具体公式如下：

符号代表意义：

Vui表示的是第i个脚本使用的并发用户数

Rtj表示的是第i个脚本第j个事务花费的时间，此时间会影响整个Action时间

Rti表示的是第i个脚本一次完成所有操作的时间，即Action时间

n 表示的是第n个脚本

m 表示的是每个脚本中m个事务

那么第j个事务的TPS = Vui/Rti

总的TPS=

4.  如何获取Vu和TPS

Ø  并发用户数(Vu)获取

新系统：没有历史数据作参考，只能通过业务部门进行评估。

旧系统：对于已经上线的系统，可以选取高峰时刻，在一定时间内使用系统的人数，这些人数认为属于在线用户数，并发用户数取10%就可以了，例如在半个小时内，使用系统的用户数为10000，那么取10%作为并发用户数基本就够了。

Ø  TPS获取

       新系统：没有历史数据作参考，只能通过业务部门进行评估。

       旧系统：对于已经上线的系统，可以选取高峰时刻，在5分钟或10分钟内，获取系统每笔交易的业务量和总业务量，按照单位时间内完成的笔数计算出TPS，即业务笔数/单位时间（5*60或10*60）

5.  如何评价系统的性能

针对服务器端的性能，以TPS为主来衡量系统的性能，并发用户数为辅来衡量系统的性能，如果必须要用并发用户数来衡量的话，需要一个前提，那就是交易在多长时间内完成，因为在系统负载不高的情况下，将思考时间(思考时间的值等于交易响应时间)加到脚本中，并发用户数基本可以增加一倍，因此用并发用户数来衡量系统的性能没太大的意义。

6.  相关案例

通过大量性能测试我们发现不需要用上万的用户并发去进行测试，只要系统处理业务时间足够快，几百个用户甚至几十个用户就可以达到目的。另外咨询很多专家做过的性能测试项目，基本都没有超过5000用户并发。

因此对于大型系统、业务量非常高、硬件配置足够多的情况下，5000用户并发就足够了；对于中小型系统，1000用户并发就足够了。

7.  性能测试策略

做性能测试需要一套标准化流程及测试策略，并发用户数只是指标考虑的一个，在做负载测试的时候，一般都是按照梯度施压的方式去加用户数，而不是在没有预估的情况下，一次加几万个用户，，交易失败率非常高，响应时间非常长，已经超过了使用者忍受范围内，这样做没有多大的意义，这就好比“有多少钱可以干多少事”一样，需要选择相关的策略。

8.  Loadrunner VS PTS

从下图对比项可以看出，PTS比Loadrunner(LR)更能让客户接受。

9.  总结

Ø  系统的性能由TPS决定，跟并发用户数没有多大关系。在同样的TPS下，可以由不同的用户数去压（通过加思考时间设置）。

Ø  系统的最大TPS是一定的（在一个范围内），但并发用户数不一定，可以调整。

Ø  建议性能测试的时候，不要设置过长的思考时间，以最坏的情况下对服务器施压。

Ø  一般情况下，大型系统（业务量大、机器多）做压力测试，5000个用户并发就够了，中小型系统做压力测试，1000个用户并发就足够了。

首先说说我们的Tomcat7.0的目录结构：

       包括目录：
       bin

       conf

       lib

       logs

       temp

       webapps

       work

       bin文件：

                   存放各种平台下启动和关闭Tomcat的脚本文件。

                   startup.bat是windows下启动tomcat的文件。

                   shutdown.bat是关闭tomcat的文件。

      conf目录：

                   Tomcat的各种配置文件，tomcat启动时需要读取的配置文件：server.xml

      lib目录：

                 存放Tomcat所需的各种jar文件。

      logs目录：
                 存放Tomcat的日志文件。

      temp目录：

                  临时文件目录，用于存放Tomcat运行时的临时文件。

      webapps目录：

                  web应用的发布目录，把java编写的web应用或打包成的war格式的web应用文件放入该目录下，就能通过Tomcat服务器访问了。

       work目录：

                Tomcat把JSP文件编译生成Servlet文件存放在这个目录下。

数据传送流程：

java servlet：

        我们要在webapps目录下建立一个我们javaweb项目的目录结构test_servlet：

        WEB-INF/classes web.xml lib 

        lib文件：存放java包，数据库驱动程序等等。

        classes文件：存放已编译好的类。

        Web.Xml：配置一些信息，例如Servlet的配置，名字，路径等，告诉我们的Tomcat我们使用的哪个Servlet。

        我们依次来写：

        servlet代码：

//引入java的包。
import javax.servlet.http.*;
import java.io.*;
import javax.servlet.*;
//第一个Servlet程序。
public class LoginServlet extends HttpServlet
{
//指定调用哪个servlet。
//get提交和post提交两种方式。
public void doGet(HttpServletRequest request,HttpServletResponse response) throws ServletException,IOException
{
//取得客户端的数据。
String username = request.getParameter("username");
String password = request.getParameter("password");
System.out.println("username" + username);
System.out.println("password" + password);
//向浏览器上写数据。可以有流。字符流。
//setContext设置响应的内容类型。设置为html类型或者中文。
response.setContentType("text/html;charset=GB18030");
//response.getWriter().println("Login Success!!!");
//response.getWriter().println("登录成功!!!");
response.getWriter().println("<html>");
response.getWriter().println("<head>");
response.getWriter().println("<title>登录信息</title>");
response.getWriter().println("</head>");
response.getWriter().println("<body>");
response.getWriter().println("欢迎【" + username + "】登录成功！！！");
response.getWriter().println("</body>");
response.getWriter().println("</html>");
}
//采用post的提交方式。
public void doPost(HttpServletRequest request,HttpServletResponse response) throws ServletException,IOException
{
doGet(request,response);
}
}

         虽然我们引入了jar包，但是程序是否能从找到tomcat的servlet  jar包，我们需要配置一下用户的环境变量，把lib\servlet-api.jar包配置到classpath变量。当然tomcat也是java编码的，所以我们也配置一下JAVA_HOME变量，可以在tomcat的批处理命令startup.bat中进行配置，set JAVA_HOME=C:\Program Files\Java\jdk1.6.0_43。

         然后把这个java文件编译成.class 文件，放到classes文件夹中。

         html代码：

<html>
<head>
<title>登录</title>
</head>
<body>
    <!--这是绝对路径-->
<!--相对路径写法:action="LoginServlet"-->
<form action="http://127.0.0.1:8080/test_servlet/LoginServlet">
用户：<input type="text" name="username"><br>
密码：<input type="password" name="password"><br>
<input type="submit" value="登录">
</form>
</body>
</html>

         Web.Xml :

<?xml version="1.0" encoding="ISO-8859-1"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
                      http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
  version="3.0"
  metadata-complete="true">
    <servlet>
<servlet-name>MyServlet</servlet-name>
<servlet-class>LoginServlet</servlet-class>
</servlet>
<servlet-mapping>
<servlet-name>MyServlet</servlet-name>
<url-pattern>/LoginServlet</url-pattern>
</servlet-mapping>
</web-app>

         启动Tomcat就可以执行成功。

         下面是servlet的调用过程,也可以用HTTPLook软件来跟踪：

       作为一个java的初学者来说，开始一个servlet让我很模糊，什么是servlet呢？有道上说是servlet（小服务程序），学过了发现，也是26个英文字母的组合，其实说白了就是java代码，但是因为他的用途，人们给他抽象总结了一个层次，叫做servlet，。就相当于县长首先是人，再是县长~，貌似搞得这儿复杂~不也是个干活的么~

      同样是网页，.net开发的网页和java开发的有什么区别呢？.net中有没有servlet？据说.cs文件写“servlet”就相当于servlet了。呵呵，有待更深的研究，加油！

        在实现缓存排序功能之前，必须先明白这一功能的合理性。不妨思考一下，既然可以在数据库中排序，为什么还要把排序功能放在缓存中实现呢？这里简单总结了两个原因：首先，排序会增加数据库的负载，难以支撑高并发的应用；其次，在缓存中排序不会遇到表锁定的问题。Redis恰好提供了排序功能，使我们可以方便地实现缓存排序。

        Redis中用于实现排序功能的是SORT命令。该命令提供了多种参数，可以对列表，集合和有序集合进行排序。SORT命令格式如下：

SORT key [BY pattern] [LIMIT offset count] [GET pattern [GET pattern ...]] [ASC | DESC] [ALPHA] [STORE destination]

SORT resultset.hash:123456 BY *->timestamp

        LIMIT参数用于限制排序以后返回元素的数量，功能类似于SQL中的limit。该参数接受另外两个参数，即offset和count，LIMIT offset count表示跳过前offset个元素，返回之后的连续count个元素。可见，LIMIT参数可以用于实现分页功能。

        GET参数用于返回指定的字段值。以集合resultset.hash:123456为例，使用BY参数对集合中的所有哈希键按照哈希结构中的timestamp字段排序后，SORT命令返回所有排序之后的哈希键。如果某个请求需要不是键而是某些字段值，这时就要使用GET参数，使SORT命令返回指定字段值。假设除timestamp字段以外，集合中每个哈希键对应的哈希结构中还有一个名为“id”的字段，通过以下命令可以使SORT返回按照timestamp排序以后的每个哈希键对应的哈希结构中的timestamp和id值：

[plain] view plaincopyprint?

1. SORT resultset.hash:123456 BY *->timestamp GET *->timestamp GET *->id  

SORT resultset.hash:123456 BY *->timestamp GET *->timestamp GET *->id

        ASC和DESC参数用于指定排序顺序（默认为ASC，即从低到高），ALPHA参数用于按照字典顺序排列非数字元素。

        STORE参数用于将SORT命令的返回值，即排序结果存入一个指定的列表。加上STORE参数后，SORT命令的返回值就变为排序结果的个数。

        下面的代码实现了按照哈希的某个字段对集合中的哈希键排序，并将结果存入列表的过程：

[cpp] view plaincopyprint?

1. // 该函数对集合中的所有HASH键进行排序，排序依据是HASH键所对应的HASH中的某个字段，  
2. // 排序结果被存入一个LIST结构，LIST键应当包含结果集标识符和排序字段标识符，  
3. // 形如“sorted:123456:1234”  
4. string SortHash(sql::Connection *mysql_connection,  
5.                 redisContext *redis_connection,   
6.                 const string &resultset_id,   
7.                 const string &sort_field,   
8.                 int offset, int count, int order, int ttl) {  
9.   // 只考虑存储HASH键的SET  
10.   string redis_row_set_key = "resultset.hash:" + resultset_id;  
11.   redisReply *reply;  
12.   // 检测SET是否存在  
13.   reply = static_cast<redisReply*>(redisCommand(redis_connection,   
14.                                                "EXISTS %s",  
15.                                                redis_row_set_key.c_str()));  
16.   if (reply->integer == 0) {  
17.     freeReplyObject(reply);  
18.     throw runtime_error("FAILURE - no resultsets");  
19.   } else {  
20.     freeReplyObject(reply);  
21.   }  
22.   string field_md5 = md5(sort_field);  // 利用MD5排除排序字段中空格造成的影响   
23.   // 将排序结果存入该LIST  
24.   string redis_sorted_list_key = "sorted:" + resultset_id + ":" + field_md5;  
25.   string by("*->" + sort_field);  //确定排序字段  
26.   string ord = (order == 1) ? "ASC" : "DESC";  //order==1时按照升序排列；否则为降序  
27.   stringstream ofsstream, cntstream;  
28.   ofsstream << offset;  
29.   cntstream << count;  
30.   // 执行排序命令，并把排序结果存入LIST  
31.   reply = static_cast<redisReply*>(redisCommand(  
32.                                       redis_connection,   
33.                                       "SORT %s BY %s LIMIT %s %s GET %s ALPHA STORE %s",  
34.                                       redis_row_set_key.c_str(),   
35.                                       by.c_str(),   
36.                                       ofsstream.str().c_str(),   
37.                                       cntstream.str().c_str(),   
38.                                       "#",   
39.                                       redis_sorted_list_key.c_str()));  
40.   freeReplyObject(reply);  
41.   stringstream ttlstream;  
42.   ttlstream << ttl;  
43.   // 设置LIST的过期时间  
44.   reply = static_cast<redisReply*>(redisCommand(redis_connection,   
45.                                                "EXPIRE %s %s",  
46.                                    redis_sorted_list_key.c_str(),   
47.                                                ttlstream.str().c_str()));  
48.   freeReplyObject(reply);  
49.   return redis_sorted_list_key;  // 返回LIST键，以便于其他函数获取该LIST中的内容  

// 该函数对集合中的所有HASH键进行排序，排序依据是HASH键所对应的HASH中的某个字段，
// 排序结果被存入一个LIST结构，LIST键应当包含结果集标识符和排序字段标识符，
// 形如“sorted:123456:1234”
string SortHash(sql::Connection *mysql_connection,
                redisContext *redis_connection, 
                const string &resultset_id, 
                const string &sort_field, 
                int offset, int count, int order, int ttl) {
  // 只考虑存储HASH键的SET
  string redis_row_set_key = "resultset.hash:" + resultset_id;
  redisReply *reply;
  // 检测SET是否存在
  reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                               "EXISTS %s",
                                               redis_row_set_key.c_str()));
  if (reply->integer == 0) {
    freeReplyObject(reply);
    throw runtime_error("FAILURE - no resultsets");
  } else {
    freeReplyObject(reply);
  }
  string field_md5 = md5(sort_field);  // 利用MD5排除排序字段中空格造成的影响 
  // 将排序结果存入该LIST
  string redis_sorted_list_key = "sorted:" + resultset_id + ":" + field_md5;
  string by("*->" + sort_field);  //确定排序字段
  string ord = (order == 1) ? "ASC" : "DESC";  //order==1时按照升序排列；否则为降序
  stringstream ofsstream, cntstream;
  ofsstream << offset;
  cntstream << count;
  // 执行排序命令，并把排序结果存入LIST
  reply = static_cast<redisReply*>(redisCommand(
                                      redis_connection, 
                                      "SORT %s BY %s LIMIT %s %s GET %s ALPHA STORE %s",
                                      redis_row_set_key.c_str(), 
                                      by.c_str(), 
                                      ofsstream.str().c_str(), 
                                      cntstream.str().c_str(), 
                                      "#", 
                                      redis_sorted_list_key.c_str()));
  freeReplyObject(reply);
  stringstream ttlstream;
  ttlstream << ttl;
  // 设置LIST的过期时间
  reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                               "EXPIRE %s %s",
                       redis_sorted_list_key.c_str(), 
                                               ttlstream.str().c_str()));
  freeReplyObject(reply);
  return redis_sorted_list_key;  // 返回LIST键，以便于其他函数获取该LIST中的内容

[cpp] view plaincopyprint?

1. // 该函数根据sql语句和排序参数，在Redis中查询相应的结果集并进行排序，最后返回  
2. // 排序之后的HASH键  
3. vector<string> GetSortedCache(sql::Connection *mysql_connection,  
4.                               redisContext *redis_connection,  
5.                               const string &sql, const string &sort_field,   
6.                               int offset, int count, int order, int ttl) {  
7.   vector<string> redis_row_key_vector;  
8.   redisReply *reply;  
9.   string resultset_id = md5(sql);  // 结果集标识符  
10.   string field_md5 = md5(sort_field);  // 排序字段标识符  
11.   // 尝试获取LIST中的所有HASH键  
12.   string redis_sorted_list_key = "sorted:" + resultset_id + ":" + field_md5;  
13.   // 尝试获取LIST中的所有HASH键  
14.   reply = static_cast<redisReply*>(redisCommand(redis_connection,   
15.                                                "LRANGE %s %s %s",  
16.                                                redis_sorted_list_key.c_str(),   
17.                                                "0",   
18.                                                "-1"));  
19.   if (reply->type == REDIS_REPLY_ARRAY) {  
20.     // 如果LIST不存在，调用Cache2Hash函数从Mysql中拉取数据到Redis，然后调用SortHash函数  
21.     // 对结果集进行排序并将排序后的HASH键存入LIST  
22.     if (reply->elements == 0) {   
23.       freeReplyObject(reply);  
24.       sql::Statement *stmt = mysql_connection->createStatement();  
25.       sql::ResultSet *resultset = stmt->executeQuery(sql);  
26.       Cache2Hash(mysql_connection, redis_connection, resultset,   
27.                  resultset_id, ttl);  
28.       redis_sorted_list_key = SortHash(mysql_connection, redis_connection,   
29.                                        resultset_id, sort_field, offset,   
30.                                        count, order, ttl);  
31.       // 再次尝试获取LIST中的所有HASH键  
32.       reply = static_cast<redisReply*>(redisCommand(  
33.                                           redis_connection,   
34.                                           "LRANGE %s %s %s",  
35.                                           redis_sorted_list_key.c_str(),   
36.                                           "0",   
37.                                           "-1"));  
38.       delete resultset;  
39.       delete stmt;  
40.     }  
41.     // 将LIST中的所有HASH键存入redis_row_key_vector中  
42.     string redis_row_key;  
43.     for (int i = 0; i < reply->elements; ++i) {  
44.       redis_row_key = reply->element[i]->str;  
45.       redis_row_key_vector.push_back(redis_row_key);  
46.     }  
47.     freeReplyObject(reply);  
48.   } else {  
49.     freeReplyObject(reply);  
50.     throw runtime_error("FAILURE - LRANGE error");  
51.   }  
52.   return redis_row_key_vector;  
53. }  

// 该函数根据sql语句和排序参数，在Redis中查询相应的结果集并进行排序，最后返回
// 排序之后的HASH键
vector<string> GetSortedCache(sql::Connection *mysql_connection,
                              redisContext *redis_connection,
                              const string &sql, const string &sort_field, 
                              int offset, int count, int order, int ttl) {
  vector<string> redis_row_key_vector;
  redisReply *reply;
  string resultset_id = md5(sql);  // 结果集标识符
  string field_md5 = md5(sort_field);  // 排序字段标识符
  // 尝试获取LIST中的所有HASH键
  string redis_sorted_list_key = "sorted:" + resultset_id + ":" + field_md5;
  // 尝试获取LIST中的所有HASH键
  reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                               "LRANGE %s %s %s",
                                               redis_sorted_list_key.c_str(), 
                                               "0", 
                                               "-1"));
  if (reply->type == REDIS_REPLY_ARRAY) {
    // 如果LIST不存在，调用Cache2Hash函数从Mysql中拉取数据到Redis，然后调用SortHash函数
    // 对结果集进行排序并将排序后的HASH键存入LIST
    if (reply->elements == 0) { 
      freeReplyObject(reply);
      sql::Statement *stmt = mysql_connection->createStatement();
      sql::ResultSet *resultset = stmt->executeQuery(sql);
      Cache2Hash(mysql_connection, redis_connection, resultset, 
                 resultset_id, ttl);
      redis_sorted_list_key = SortHash(mysql_connection, redis_connection, 
                                       resultset_id, sort_field, offset, 
                                       count, order, ttl);
      // 再次尝试获取LIST中的所有HASH键
      reply = static_cast<redisReply*>(redisCommand(
                                          redis_connection, 
                                          "LRANGE %s %s %s",
                                          redis_sorted_list_key.c_str(), 
                                          "0", 
                                          "-1"));
      delete resultset;
      delete stmt;
    }
    // 将LIST中的所有HASH键存入redis_row_key_vector中
    string redis_row_key;
    for (int i = 0; i < reply->elements; ++i) {
      redis_row_key = reply->element[i]->str;
      redis_row_key_vector.push_back(redis_row_key);
    }
    freeReplyObject(reply);
  } else {
    freeReplyObject(reply);
    throw runtime_error("FAILURE - LRANGE error");
  }
  return redis_row_key_vector;
}

        把Mysql结果集缓存到Redis的字符串或哈希结构中以后，我们面临一个新的问题，即如何为这些字符串或哈希命名，也就是如何确定它们的键。因为这些数据结构所对应的行都属于某个结果集，假如可以找到一种唯一标识结果集的方法，那么只需为这些数据结构分配一个唯一的序号，然后把结果集标识符与该序号结合起来，就能唯一标识一个数据结构了。于是，为字符串和哈希命名的问题就转化为确定结果集标识符的问题。

        经过调研，发现一种较为通用的确定结果集标识符的方法。正如我们所知道的，缓存在Redis中的结果集数据都是利用select等sql语句从Mysql中获取的。同样的查询语句会生成同样的结果集（这里暂时不讨论结果集中每条记录的顺序问题），这一性质刚好可以用来确定结果集的唯一标识符。当然，简单地把整个sql语句作为结果集标识符是不可取的，一个显而易见的理由是，未经处理的sql查询语句均包含若干空格，而Redis的键是不允许存在空格的。这时，我们需要一个可以把sql语句转换为唯一标识符的函数。通常，这一功能由散列函数完成，包括MD5，SHA系列等加密散列函数在内的很多算法均可达到这一目的。

        确定结果集标识符之后，从Redis读数据或向Redis写数据的思路就很清晰了。对于一个sql语句格式的数据请求，首先计算该语句的MD5并据此得到结果集标识符，然后利用该标识符在Redis中查找该结果集。注意，结果集中的每一行都有一个相应的键，这些键都存储在一个Redis集合结构中。这个集合恰好对应了所需的结果集，所以，该集合的键必须包含结果集标识符。如果Redis中不存在这样一个集合，说明要找的结果集不在Redis中，所以需要执行相应的sql语句，在Mysql中查询到相应的结果集，然后按照上面所说的办法把结果集中的每一行以字符串或哈希的形式存入Redis。在Redis中查找相应结果集的代码如下：

// 该函数根据sql语句在Redis中查询相应的结果集，并返回结果集中每一行所对应的数据结构的键
vector<string> GetCache(sql::Connection *mysql_connection,
                      redisContext *redis_connection,
                      const string &sql, int ttl, int type) {
  vector<string> redis_row_key_vector;
  string resultset_id = md5(sql);  // 计算sql语句的md5，这是唯一标识结果集的关键
  // type==1时，该函数将查询相应的STRING集合或将结果集写入若干STRING
  string cache_type = (type == 1) ? "string" : "hash";
  // 根据type信息和结果集标识符合成SET键
  string redis_row_set_key = "resultset." + cache_type + ":" + resultset_id;
  redisReply *reply;
  // 尝试从reply中获取SET中保存的所有键
  reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                               "SMEMBERS %s",
                                               redis_row_set_key.c_str()));
  if (reply->type == REDIS_REPLY_ARRAY) {
// 如果要找的SET不存在，说明Redis中没有相应的结果集，需要调用Cache2String或
// Cache2Hash函数把数据从Mysql拉取到Redis中
    if (reply->elements == 0) {
      freeReplyObject(reply);
      sql::Statement *stmt = mysql_connection->createStatement();
      sql::ResultSet *resultset = stmt->executeQuery(sql);
      if (type == 1) {
        redis_row_set_key = Cache2String(mysql_connection, redis_connection,
                                         resultset, resultset_id, ttl);
      } else {
        redis_row_set_key = Cache2Hash(mysql_connection, redis_connection, 
                                       resultset, resultset_id, ttl);
      }
  // 再次尝试从reply中获取SET中保存的所有键
      reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                                   "SMEMBERS %s",
                                                   redis_row_set_key.c_str()));
      delete resultset;
      delete stmt;
    }
// 把SET中的每个STRING或HASH键存入redis_row_key_vector中
    string redis_row_key;
    for (int i = 0; i < reply->elements; ++i) {
      redis_row_key = reply->element[i]->str;
      redis_row_key_vector.push_back(redis_row_key);
    }
    freeReplyObject(reply);
  } else {
    freeReplyObject(reply);
    throw runtime_error("FAILURE - SMEMBERS error");
  }
  return redis_row_key_vector;
}

        用Redis作Mysql数据库缓存，必须解决2个问题。首先，应该确定用何种数据结构存储来自Mysql的数据；在确定数据结构之后，还要考虑用什么标识作为该数据结构的键。

        直观上看，Mysql中的数据都是按表存储的；更微观地看，这些表都是按行存储的。每执行一次select查询，Mysql都会返回一个结果集，这个结果集由若干行组成。所以，一个自然而然的想法就是在Redis中找到一种对应于Mysql行的数据结构。Redis中提供了五种基本数据结构，即字符串（string）、列表（list）、哈希（hash）、集合（set）和有序集合（sorted set）。经过调研，发现适合存储行的数据结构有两种，即string和hash。

        要把Mysql的行数据存入string，首先需要对行数据进行格式化。事实上，结果集的每一行都可以看做若干由字段名和其对应值组成的键值对集合。这种键值对结构很容易让我们想起Json格式。因此，这里选用Json格式作为结果集每一行的格式化模板。根据这一想法，我们可以实现将结果集格式化为若干Json对象，并将Json对象转化为字符串存入Redis的代码：

[cpp] view plaincopyprint?

1. // 该函数把结果集中的每一行转换为一个Json格式的字符串并存入Redis的STRING结构中，  
2. // STRING键应该包含结果集标识符和STRING编号，形式如“cache.string:123456:1”  
3. string Cache2String(sql::Connection *mysql_connection,  
4.                     redisContext *redis_connection,  
5.                     sql::ResultSet *resultset,  
6.                     const string &resultset_id, int ttl) {  
7.   if (resultset->rowsCount() == 0) {  
8.     throw runtime_error("FAILURE - no rows");  
9.   }  
10.   // STRING键的前缀，包含了结果集的标识符  
11.   string prefix("cache.string:" + resultset_id + ":");  
12.   unsigned int num_row = 1;  // STRING编号，附加于STRING键的末尾，从1开始  
13.   sql::ResultSetMetaData *meta = resultset->getMetaData();  
14.   unsigned int num_col = meta->getColumnCount();  
15.   // 将结果集中所有行对应的所有STRING键存入该SET，SET键包含了结果集的标识符  
16.   string redis_row_set_key("resultset.string:" + resultset_id);  
17.   redisReply *reply;  
18.   string ttlstr;  
19.   stringstream ttlstream;  
20.   ttlstream << ttl;  
21.   ttlstr = ttlstream.str();  
22.   resultset->beforeFirst();  
23.   // 将结果集中的每一行转为Json格式的字符串，将这些Json字符串存入STRING，  
24.   // 每个STRING对应结果集中的一行  
25.   while (resultset->next()) {  
26.     string redis_row_key;  // STRING键名，由前缀和STRING编号组成  
27.     stringstream keystream;  
28.     keystream << prefix << num_row;  
29.     redis_row_key = keystream.str();  
30.     Json::Value row;  
31.     for (int i = 1; i <= num_col; ++i) {  
32.       string col_label = meta->getColumnLabel(i);  
33.       string col_value = resultset->getString(col_label);  
34.       row[col_label] = col_value;  
35.     }  
36.     Json::FastWriter writer;  
37.     string redis_row_value = writer.write(row);  
38.     // 将STRING键及Json格式的对应值对存入Redis  
39.     reply = static_cast<redisReply*>(redisCommand(redis_connection,   
40.                                                  "SET %s %s",  
41.                                                  redis_row_key.c_str(),   
42.                                                  redis_row_value.c_str()));  
43.     freeReplyObject(reply);  
44.     // 将STRING键加入SET中  
45.     reply = static_cast<redisReply*>(redisCommand(redis_connection,   
46.                                                  "SADD %s %s",  
47.                                                  redis_row_set_key.c_str(),   
48.                                                  redis_row_key.c_str()));  
49.     freeReplyObject(reply);  
50.     // 设置STRING的过期时间  
51.     reply = static_cast<redisReply*>(redisCommand(redis_connection,   
52.                                                  "EXPIRE %s %s",  
53.                                                  redis_row_key.c_str(),   
54.                                                  ttlstr.c_str()));  
55.     freeReplyObject(reply);  
56.     ++num_row;  
57.   }  
58.   // 设置SET的过期时间  
59.   reply = static_cast<redisReply*>(redisCommand(redis_connection,   
60.                                                "EXPIRE %s %s",  
61.                                                redis_row_set_key.c_str(),   
62.                                                ttlstr.c_str()));  
63.   freeReplyObject(reply);  
64.   return redis_row_set_key;  // 返回SET键，以便于其他函数获取该SET中的内容  
65. }  

// 该函数把结果集中的每一行转换为一个Json格式的字符串并存入Redis的STRING结构中，
// STRING键应该包含结果集标识符和STRING编号，形式如“cache.string:123456:1”
string Cache2String(sql::Connection *mysql_connection,
                    redisContext *redis_connection,
                    sql::ResultSet *resultset,
                    const string &resultset_id, int ttl) {
  if (resultset->rowsCount() == 0) {
    throw runtime_error("FAILURE - no rows");
  }
  // STRING键的前缀，包含了结果集的标识符
  string prefix("cache.string:" + resultset_id + ":");
  unsigned int num_row = 1;  // STRING编号，附加于STRING键的末尾，从1开始
  sql::ResultSetMetaData *meta = resultset->getMetaData();
  unsigned int num_col = meta->getColumnCount();
  // 将结果集中所有行对应的所有STRING键存入该SET，SET键包含了结果集的标识符
  string redis_row_set_key("resultset.string:" + resultset_id);
  redisReply *reply;
  string ttlstr;
  stringstream ttlstream;
  ttlstream << ttl;
  ttlstr = ttlstream.str();
  resultset->beforeFirst();
  // 将结果集中的每一行转为Json格式的字符串，将这些Json字符串存入STRING，
  // 每个STRING对应结果集中的一行
  while (resultset->next()) {
    string redis_row_key;  // STRING键名，由前缀和STRING编号组成
    stringstream keystream;
    keystream << prefix << num_row;
    redis_row_key = keystream.str();
    Json::Value row;
    for (int i = 1; i <= num_col; ++i) {
      string col_label = meta->getColumnLabel(i);
      string col_value = resultset->getString(col_label);
      row[col_label] = col_value;
    }
    Json::FastWriter writer;
    string redis_row_value = writer.write(row);
// 将STRING键及Json格式的对应值对存入Redis
    reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                                 "SET %s %s",
                                                 redis_row_key.c_str(), 
                                                 redis_row_value.c_str()));
    freeReplyObject(reply);
    // 将STRING键加入SET中
    reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                                 "SADD %s %s",
                                                 redis_row_set_key.c_str(), 
                                                 redis_row_key.c_str()));
    freeReplyObject(reply);
    // 设置STRING的过期时间
    reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                                 "EXPIRE %s %s",
                                                 redis_row_key.c_str(), 
                                                 ttlstr.c_str()));
    freeReplyObject(reply);
    ++num_row;
  }
  // 设置SET的过期时间
  reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                               "EXPIRE %s %s",
                                               redis_row_set_key.c_str(), 
                                               ttlstr.c_str()));
  freeReplyObject(reply);
  return redis_row_set_key;  // 返回SET键，以便于其他函数获取该SET中的内容
}

        要把Mysql的行数据存入hash，过程要比把数据存入string直观很多。这是由hash的结构性质决定的——hash本身就是一个键值对集合：一个“父键”下面包含了很多“子键”，每个“子键”都对应一个值。根据前面的分析可知，结果集中的每一行实际上也是键值对集合。用Redis键值对集合表示Mysql键值对集合应该再合适不过了：对于结果集中的某一行，字段对应于hash的“子键”，字段对应的值就是hash“子键”对应的值，即结果集的一行刚好对应一个hash。这一想法的实现代码如下：

[cpp] view plaincopyprint?

1. // 该函数把结果集中的每一行都存入一个HASH结构。HASH键应当包括结果集标识符和HASH编号，  
2. // 形如“cache.string:123456:1”  
3. string Cache2Hash(sql::Connection *mysql_connection,  
4.                   redisContext *redis_connection,  
5.                   sql::ResultSet *resultset,  
6.                   const string &resultset_id, int ttl) {  
7.   if (resultset->rowsCount() == 0) {  
8.     throw runtime_error("FAILURE - no rows");  
9.   }  
10.   // HASH键的前缀，包含了结果集的标识符  
11.   string prefix("cache.hash:" + resultset_id + ":");  
12.   unsigned int num_row = 1;  // HASH编号，附加于HASH键的末尾，从1开始  
13.   sql::ResultSetMetaData *meta = resultset->getMetaData();  
14.   unsigned int num_col = meta->getColumnCount();  
15.   // 将结果集中所有行对应的所有HASH键存入该SET，SET键包含了结果集的标识符  
16.   string redis_row_set_key("resultset.hash:" + resultset_id);  
17.   redisReply *reply;  
18.   string ttlstr;  
19.   stringstream ttlstream;  
20.   ttlstream << ttl;  
21.   ttlstr = ttlstream.str();  
22.   // 结果集中的每一行对应于一个HASH，将结果集的所有行都存入相应HASH中  
23.   resultset->beforeFirst();  
24.   while (resultset->next()) {  
25.     string redis_row_key;  // HASH键名，由前缀和HASH编号组成  
26.     stringstream keystream;  
27.     keystream << prefix << num_row;  
28.     redis_row_key = keystream.str();  
29.     for (int i = 1; i <= num_col; ++i) {  
30.       string col_label = meta->getColumnLabel(i);  
31.       string col_value = resultset->getString(col_label);  
32.       // 将结果集中一行的字段名和对应值存入HASH  
33.       reply = static_cast<redisReply*>(redisCommand(redis_connection,  
34.                                                    "HSET %s %s %s",  
35.                                                    redis_row_key.c_str(),   
36.                                                    col_label.c_str(),  
37.                                                    col_value.c_str()));  
38.       freeReplyObject(reply);  
39.     }  
40.     // 将HASH键加入SET中  
41.     reply = static_cast<redisReply*>(redisCommand(redis_connection,   
42.                                                  "SADD %s %s",  
43.                                                  redis_row_set_key.c_str(),   
44.                                                  redis_row_key.c_str()));   
45.     freeReplyObject(reply);  
46.     // 设置HASH的过期时间  
47.     reply = static_cast<redisReply*>(redisCommand(redis_connection,   
48.                                                  "EXPIRE %s %s",  
49.                                                  redis_row_key.c_str(),   
50.                                                  ttlstr.c_str()));  
51.     freeReplyObject(reply);  
52.     ++num_row;  
53.   }  
54.   // 设置SET的过期时间  
55.   reply = static_cast<redisReply*>(redisCommand(redis_connection,   
56.                                                "EXPIRE %s %s",  
57.                                                redis_row_set_key.c_str(),   
58.                                                ttlstr.c_str()));  
59.   freeReplyObject(reply);  
60.   return redis_row_set_key;  // 返回SET键，以便于其他函数获取该SET中的内容  
61. }  

// 该函数把结果集中的每一行都存入一个HASH结构。HASH键应当包括结果集标识符和HASH编号，
// 形如“cache.string:123456:1”
string Cache2Hash(sql::Connection *mysql_connection,
                  redisContext *redis_connection,
                  sql::ResultSet *resultset,
                  const string &resultset_id, int ttl) {
  if (resultset->rowsCount() == 0) {
    throw runtime_error("FAILURE - no rows");
  }
  // HASH键的前缀，包含了结果集的标识符
  string prefix("cache.hash:" + resultset_id + ":");
  unsigned int num_row = 1;  // HASH编号，附加于HASH键的末尾，从1开始
  sql::ResultSetMetaData *meta = resultset->getMetaData();
  unsigned int num_col = meta->getColumnCount();
  // 将结果集中所有行对应的所有HASH键存入该SET，SET键包含了结果集的标识符
  string redis_row_set_key("resultset.hash:" + resultset_id);
  redisReply *reply;
  string ttlstr;
  stringstream ttlstream;
  ttlstream << ttl;
  ttlstr = ttlstream.str();
  // 结果集中的每一行对应于一个HASH，将结果集的所有行都存入相应HASH中
  resultset->beforeFirst();
  while (resultset->next()) {
    string redis_row_key;  // HASH键名，由前缀和HASH编号组成
    stringstream keystream;
    keystream << prefix << num_row;
    redis_row_key = keystream.str();
    for (int i = 1; i <= num_col; ++i) {
      string col_label = meta->getColumnLabel(i);
      string col_value = resultset->getString(col_label);
  // 将结果集中一行的字段名和对应值存入HASH
      reply = static_cast<redisReply*>(redisCommand(redis_connection,
                                                   "HSET %s %s %s",
                                                   redis_row_key.c_str(), 
                                                   col_label.c_str(),
                                                   col_value.c_str()));
      freeReplyObject(reply);
    }
// 将HASH键加入SET中
    reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                                 "SADD %s %s",
                                                 redis_row_set_key.c_str(), 
                                                 redis_row_key.c_str())); 
    freeReplyObject(reply);
// 设置HASH的过期时间
    reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                                 "EXPIRE %s %s",
                                                 redis_row_key.c_str(), 
                                                 ttlstr.c_str()));
    freeReplyObject(reply);
    ++num_row;
  }
  // 设置SET的过期时间
  reply = static_cast<redisReply*>(redisCommand(redis_connection, 
                                               "EXPIRE %s %s",
                                               redis_row_set_key.c_str(), 
                                               ttlstr.c_str()));
  freeReplyObject(reply);
  return redis_row_set_key;  // 返回SET键，以便于其他函数获取该SET中的内容
}

        作为Redis提供的最简单的数据结构，字符串示意图及其常用命令如下图所示：

图1

        可见，Redis字符串类型实际上类似于C++中的map，一个键对应一个值。列表类型则是一种相对高级的数据结构，其示意图和常用命令如下图所示：

图2

        列表实际上由一个双向链表实现，所以在列表两端插入或删除数据效率极高，可以用于实现类似于微博、人人网上的“新鲜事”功能。 Redis提供的一个更有趣的结构称为哈希，其示意图和常用命令如下图所示：

图3

        一个哈希实际上是一组键值对的集合，哈希本身的键是“父键”，哈希中包含的若干字段是“子键”，每个“子键”都有相应的值。“子键”之间并没有一定的联系，但是它们共同组成了一个完整的哈希结构。我们可以把一个哈希看做关系数据库中的一行，哈希的每个子键对应行的一个字段。因此，当把关系数据库中的数据缓存至Redis时，使用哈希结构可能会带来方便。需要注意的是，哈希结构内部的子键之间是没有顺序关系的。Redis提供的另一个无顺序关系的结构是集合，其示意图和常用命令如下图所示：

图4

        集合中的数据满足唯一性和无序性。集合在Redis内部是用哈希表实现的，所以插入和删除操作的时间复杂度均为O(1)。Redis为集合提供了求交并差等运算命令，使用起来非常方便。与集合密切相关的另一个数据结构是有序集合，其示意图和常用命令如下图所示：

图5

        有序集合与集合的最大差异在于有序集合中的每个数据都有一个对应的分数，集合中的元素就是按照这些分数进行排序的。从结构功能和使用方法上来看，有序集合可能是Redis提供的5种数据结构中最高级的一种。

        Redis是一种内存数据库，这使其在数据存取效率方面表现突出。由于内存中的数据时刻面临丢失的危险，Redis提供了两种持久化机制，及时将内存中的数据写入硬盘。第一种机制为RDB，利用存储快照的方式把内存数据定期写入硬盘；第二种机制为AOF，利用写日志的方式，每执行一条更改Redis数据的命令，就在日志里附加上该命令并保存在硬盘上。Redis默认开启RDB模式，关闭AOF模式，这是因为AOF更加耗时。但是，如果对数据安全要求极高，无法承担任何数据丢失的后果，AOF模式就变成了持久化的首选。

        当然，即使Redis具备持久化机制，但是一旦本地硬盘损坏，数据丢失依然在所难免。所以，Redis又提供了复制功能，作用是将一个主数据库（master）的数据自动同步到多个从数据库（slave），从而尽可能防止数据丢失。Redis主从机制可以简单描述为：从数据库启动后，先向主数据库发送SYNC命令；主数据库接到SYNC命令后就开始保存快照，在此期间，所有发给主数据库的命令都被缓存起来；快照保存完成后，主数据库把快照和缓存的命令一起发给从数据库；从数据库保存主数据库发来的快照文件，并依次执行主数据库发来的缓存命令。在同步过程中，从数据库不会阻塞，它默认使用同步之前的数据继续响应客户端发来的命令。

        事务是Redis提供的特性之一。首先利用MULTI命令通知Redis，接下来的若干命令属于同一事务；然后输入若干命令，这些命令都被放入命令队列而不会被立即执行；最后，利用EXEC命令通知Redis，属于同一事务的所有命令均已输入完成，现在开始执行该事物。属于同一个事务的所有有效命令或者全部被执行，或者全部不执行，且在执行过程中不会插入其他命令。管道（pipeline）是Redis提供的另一个特性。该特性使Redis能够一次性接收多个命令，执行之后再一次性返回结果。这样能够减少客户端与Redis服务器的通信次数，从而降低往返时延。

        除事务和管道以外，Redis还提供了很多其他特性，如设置键的过期时间，利用BLPOP/BRPOP命令实现优先级队列，利用PUBLISH/SUBSCRIBE命令实现消息订阅和发布等，这里不再赘述。

        Redis由ANSI C写成，能够运行在包括Linux和OS X在内的大部分POSIX系统上，官方推荐的操作系统为Linux。Redis没有官方的Windows版本，但是微软移植并维护了一个能够运行在Windows上的Redis并放在github上（https://github.com/MSOpenTech/redis）。

        Redis的编程接口被称为客户端（clients），绝大部分主流编程语言都有官方推荐的客户端。下图中，具有Redis官方推荐客户端的语言标为绿色，没有官方推荐客户端的标为蓝色：

图6

        由于Redis没有官方推荐的C++客户端，因此可以使用C客户端予以代替。官方推荐的C客户端是hiredis，可以从github上找到（https://github.com/redis/hiredis）。

        作为一款功能强大，效率极高的NoSQL数据库，Redis已被包括新浪微博，知乎，Stackoverflow，暴雪和Flickr在内的多家公司成功用于实战。

注：图1-5均来自《Redis in Action》,Josiah L.Carlson, Manning Publications, 2013

参考文献：

1. 李子骅，Redis入门指南，人民邮电出版社，2013年5月第1版

2.Josiah L.Carlson, Redis in Action, Manning Publications, 2013

3. Redis官方网站http://www.redis.io

1 技术问题

• 系统 目前常见的Unix/Linux系统都可满足php(Apache)、java的运行，其中大部分默认支持System V消息队列。

• php php需要通过消息队列和java进程通信，按php的说明，php在4.3.0版本以后支持System V消息队列。

• apache 无特殊要求，满足php要求即可。

• java java版本在1.5以后。

• 在Unix/Linux环境中，推荐使用消息队列模式。

• 系统 没有限制，很难找到不支持TCP/IP的系统。

• php 按php的说明，php版本>=4.1.0支持socket

• apache 无特殊要求，满足php要求即可。

• java java版本在1.5以后。

• Windows系统只能使用socket模式

 

Dubbo是一个分布式服务框架，致力于提供高性能和透明化的RPC远程服务调用方案，以及SOA服务治理方案。简单的说，dubbo就是个服务框架，如果没有分布式的需求，其实是不需要用的，只有在分布式的时候，才有dubbo这样的分布式服务框架的需求，并且本质上是个服务调用的东东，说白了就是个远程服务调用的分布式框架（告别Web Service模式中的WSdl，以服务者与消费者的方式在dubbo上注册）
其核心部分包含:
1. 远程通讯: 提供对多种基于长连接的NIO框架抽象封装，包括多种线程模型，序列化，以及“请求-响应”模式的信息交换方式。
2. 集群容错: 提供基于接口方法的透明远程过程调用，包括多协议支持，以及软负载均衡，失败容错，地址路由，动态配置等集群支持。
3. 自动发现: 基于注册中心目录服务，使服务消费方能动态的查找服务提供方，使地址透明，使服务提供方可以平滑增加或减少机器。

2. Dubbo能做什么？

1.透明化的远程方法调用，就像调用本地方法一样调用远程方法，只需简单配置，没有任何API侵入。      
2.软负载均衡及容错机制，可在内网替代F5等硬件负载均衡器，降低成本，减少单点。
3. 服务自动注册与发现，不再需要写死服务提供方地址，注册中心基于接口名查询服务提供者的IP地址，并且能够平滑添加或删除服务提供者。

Dubbo采用全Spring配置方式，透明化接入应用，对应用没有任何API侵入，只需用Spring加载Dubbo的配置即可，Dubbo基于Spring的Schema扩展进行加载。

之前使用Web Service，我想测试接口可以通过模拟消息的方式通过soapui或LR进行功能测试或性能测试。但现在使用Dubbo，接口之间不能直接交互，我尝试通过模拟消费者地址测试，结果不堪入目，再而使用jmeter通过junit进行测试，但还是需要往dubbo上去注册，如果再不给提供源代码的前提下，这个测试用例不好写啊....

3. dubbo的架构

节点角色说明：

       Provider: 暴露服务的服务提供方。

       Consumer: 调用远程服务的服务消费方。

       Registry: 服务注册与发现的注册中心。

       Monitor: 统计服务的调用次调和调用时间的监控中心。

       Container: 服务运行容器。

这点我觉得非常好，角色分明，可以根据每个节点角色的状态来确定该服务是否正常。

调用关系说明：

0 服务容器负责启动，加载，运行服务提供者。

1. 服务提供者在启动时，向注册中心注册自己提供的服务。

2. 服务消费者在启动时，向注册中心订阅自己所需的服务。

3. 注册中心返回服务提供者地址列表给消费者，如果有变更，注册中心将基于长连接推送变更数据给消费者。

4. 服务消费者，从提供者地址列表中，基于软负载均衡算法，选一台提供者进行调用，如果调用失败，再选另一台调用。

5. 服务消费者和提供者，在内存中累计调用次数和调用时间，定时每分钟发送一次统计数据到监控中心。

dubbo的容错性显而易见，性能方面还没有还得及测，我们系统某页面需要掉5次接口，本来想建议做个缓存，但业务关系不能采纳，还需要研究下dubbo的性能调优问题...

4. dubbo使用方法。

Dubbo采用全Spring配置方式，透明化接入应用，对应用没有任何API侵入，只需用Spring加载Dubbo的配置即可，Dubbo基于Spring的Schema扩展进行加载。如果不想使用Spring配置，而希望通过API的方式进行调用（不推荐）

下面我们就来看看spring配置方式的写法:

服务提供者：

1. 下载zookeeper注册中心，下载地址：http://www.apache.org/dyn/closer.cgi/zookeeper/  下载后解压即可，进入D:\apach-zookeeper-3.4.5\bin，

双击zkServer.cmd启动注册中心服务。

2. 定义服务接口: (该接口需单独打包，在服务提供方和消费方共享)

 下面这个例子不错，写的很详细可以做个model.

在服务提供方实现接口：(对服务消费方隐藏实现)

调用结果为：

dubbo管理页面：

这个管理页面还需要部署一个环境的，一开始我还以为是dubbo自带的，找了半天没有找到....

应用页面：

提供者页面:

消费者页面：

服务页面：

测试是否成功，我觉得只要看看状态是否正常，就ok了 ....

案例代码下载：http://download.csdn.net/detail/yiyu1/7116319

转自：http://tieba.baidu.com/f?kz=692446591

怎么学习Java，这是很多新手经常会问我的问题，现在我简单描述下一个Java初学者到就业要学到的一些东西： 
    首先要明白Java体系设计到得三个方面：J2SE,J2EE,J2ME（KJAVA）。J2SE，Java 2 Platform Standard Edition，我们经常说到的JDK，就主要指的这个，它是三者的基础，属于桌面级应用开发，这部分如果学得好很容易拓展J2EE和J2ME。J2ME，The Micro Edition of the Java 2 Platform。主要用于嵌入式Java，如手机，PDA等等。J2EE，Java 2 Platform，Enterprise Edition，就是所谓的企业级Java，适合企业的大型应用开发。 
J2SE： 
    新手最初学习的时候先要从J2SE开始学起，所包含的内容主要分为：面向对象概念的理解、Java基本语法的学习，Java桌面图形界面应用程序的开发，掌握常用的Java API等（关键是要学会怎样查阅）。 
重点：Java基本语法（循环结构，数据类型，数组，运算符等）、Swing，awt，事件机制、文件输入输出流处理等
难点：面向对象思想的理解（接口、类等）、线程、socket网络编程等 
视频教程推荐：孙鑫的Java基础视频教程、张孝祥的Java基础视频教程，讲的都很细，而且这两位老师在教学方面都有很好的经验。 
J2EE： 
        在学习了J2SE之后，你可以尝试着慢慢过渡到J2EE，当然，学习J2EE海需要很多非JAVA技术的支撑，比如数据库技术，网页编程技术等等，待会我会详细介绍这两方面的内容。J2EE有三个必学的框架，如果你说你学了J2EE，但又不会这三个框架的话，那会让人笑掉大牙，这三个框架分别是Struts、Spring和Hibernate，虽然这三个框架应用很广，但不代表这三个框架就代表了J2EE，J2EE还有很多其他的东西，比如EJB，作为一款重量级（此重量级不是指重量级选手那个意思，此重量级非彼重量级）框架，虽然这个应用慢慢的再被其他框架所取代，但EJB3.0的出现也给它带回一些生机，作为一个分布式应用的框架，也是大家要去学习和了解的知识。 
        当然，刚才说到J2EE包含的范围很广，即使我把我所了解的所有技术说出来，也并不能代表J2EE，这是一个很深很广的学问，需要大家以后再工作的时候慢慢去发现了。我所了解的还包括： 
JDBC:Java数据库连接对象，基础中的基础，Hibernate也只是对它的封装而已 
JNDI: Java 命名与目录接口，J2EE重要规范之一 
EJBS: 上面已经提到过了 
RMI: 提供远程调用方法的支持，主要用于程序分布式开发 
JMS: Java消息服务，中间件技术 
JAVA IDL: 提供J2EE平台与CORBA交互能力和互联能力的技术 
JTS: 组件事务监视器 
JTA: 事务划分的一个技术 
JAVAMAIL: Java中的邮件技术 
JAF: Java启动框架,没研究过 
Log4j，一款日志处理的框架应用 
Junit：单元测试的好帮手 
freemarker、velocity：两款不错的模板引擎 
与Flash的交互Flex：目前很多的SNS应用就是使用到了这个技术 
Web services:想知道我们论坛天气预报的效果是怎么做的吗？请关注这个技术 
SOA:面向服务架构，未来技术发展的趋势之一 
…… 
        不过前面介绍到得这些都不是J2EE的基础,是属于框架应用级别的.要真正把上面说到的知识掌握好,还是的学习J2EE最核心的东西:JDBC、JSP、servlet和JavaBean,后面三个也是学习SSH的基础(Struts/Spring/Hibernate的简称).鉴于J2EE的内容太过复杂,建议大家最初只学习: JSP, servlet和JavaBean，等这个基础掌握好再把SSH学习了，再对其他J2EE技术有个大致的了解就可以出去找工作了。当然，找工作的前提除了把Java学习好，还会有一些其他的技术也是必须要掌握的。

-_- 我很简单,Java却要求我学这么多复杂的东西。除了这些技术的东西，在这阶段，大家更多的是会接触到比如设计模式的东西，比如工厂模式、桥梁模式、单例模式等等，还要学习建模思想（比如UML建模等等）和软件生命周期等等。这些是升华大家编程思想的手段和途径。 
数据库技术: 
     数据库几乎是程序中不可或缺的一部分,在整个软件开发中更是占据了重要作用.数据库在整个软件中扮演的是地基的角色,你要修房子,就要先打好地基,你要做软件,就要先设计好的数据库.目前数据库几乎都是关系型数据库,代表的有: 
SQL Server,目前最新的好像是2008,我不是很喜欢追赶时髦,用的还是2005,不过也很不错了,在此之前还有2000. 
MySQL,最知名的开源数据库之一,我很喜欢它的小巧耐用,学过SQL SERVER数据库再学它,无师自通,易如反掌. 
ORACLE.甲骨文公司的产品,大型数据库,正版的一般公司买不起.只有看看了,不过现在Java开发60%都是会使用到这个数据库,再加上现在甲骨文已经把sun公司收购了,JAVA和ORACLE得关系更加暧昧,各位得多花些时间在这上面了. 
除了这些，你还应该了解一些桌面应用软件开发用到的数据库，比如说sqlite。你在制作安装程序的时候可以把这个数据库打包进你的安装程序。 
网页编程基础: 
     网页编程基础要学的内容也比较多,但主要集中在以下这三个技术之上,大家每天看的各种样式的网页都是它们呈现的: 
HTML:超文本标记语言 
CSS:层叠样式表 
Javascript:浏览器客户端脚本语言 
现在更是流行各种JavaScript库,比如:Jquery、ExtJs等，你要是会其中的某一个库的使用，找工作so easy。 
J2ME： 
     如果你在学习了以上技术还不过瘾，可以再学习手机程序的开发，就我个人看来，开发手机程序技术其实蛮简单的，难点在于优化，比如怎么让你写的程序尽量的节约手机资源，“代码诚可贵，手机价更高“，所以不得不比PC机多考虑很多成本和实用性问题。J2ME的学习的话，切忌不要单纯的再模拟器上测试，一定要找真机测试，真机和模拟器的效果之间差别是很大的。更多手机开发技术，有兴趣的朋友可以和我一起探讨。 
开发工具： 
    开发工具的选择主要还是看大家的习惯，我个人比较喜欢使用eclispe，功能很强大，也很好拓展其他的功能。当让初学你也可以选择JCreator，或者最好就直接用记事本。等你熟悉了Java代码之后，就可以使用JBuilder、eclispe或者netbeans这一类的工具了。 
     先说到这里先，以后大家一起完善这个帖子。尽量再给大家整理一些学习的资源在这里。也欢迎大家一起整理我文章中提到的一些技术资源和开发工具。最后，再和大家分享一句话：生命欣欣向荣，唯奋斗不息。也和大家分享一幅Java学习路线图：