分布式应用技术

本文讲一些常见的分布式应用层面的技术，其中大部分都依赖于Zookeeper，所以对zookeeper不熟悉的同学可以先看我之前写的两往篇博客Zookeeper编程（二）、Zookeeper编程（一）

锁

我们的推荐程序部署在多台服务器上，每天凌晨会去重建信息索引（索引存在Redis上）。建索引这件事情只能让一台服务器去做，其他服务器歇着，这种应用场景自然想到了分布式锁，谁抢到锁谁来建索引。我们在线上分别使用过3种分布式锁：基于zookeeper的锁，基于MySQL的锁，基于Redis的锁。MySQL lock最稳定；Redis lock使用起来最灵活，同时实时起来也最方便。所以现在我们线上的分布式锁全换成了Redis Lock。

zookeeper lock的实现原理参见链接，不再赘述。

MySQL lock的基本思想是大家都去写同一条数据库记录，谁先写上谁获取锁，删除这条记录就相当于释放了锁。整个流程看起来像这样：

begin;
select count(*) from table where lockname='xxx' for update;
if count==0:
    insert into table (lockname) values ('xxx);
    do the task which must be done once
    delete from table where lockname='xxx'
commit;

如果2个进程同时执行第一步，发现lockname不存在，于是都去添加一行记录，还都添加成功了，那岂不是2个进程都获得了锁？莫急，看见select语句后面有个”for update”吗？当where条件中不包含主键时，select … for update将会锁表，事务提交后才释放表上的锁。（for update仅适用于InnoDB）

Redis lock的实现思想跟mysql相同，不过操作起来更简单。看下面的代码

Long i = jedis.setnx(lockName, lockName);    // 若key不存在，则存储 ，并返回1
if (i == 1L) {
    // 设置key的过期时间
    if (live < 0) {
        live = DEFAULT_EXPIRE_TIME;
    }
    jedis.expire(lockName, live);
    logger.info("get redis lock " + lockName + " ,live " + live
            + " seconds");

    rect = true;        //获得锁返回true
} else { // 已存在锁
    logger.info("lockName: " + lockName
            + " locked by other business");
    rect = false;        //没有获得锁返回false
}

核心操作是redis提供的setnx()方法，它来保证并发情况下中有1个进程能写成功。另外我们还为redis的key设置了超时时间，即使你获得锁后忘记了释放锁，或者在释放锁之前进程死掉了，不用担心，在达到超时时间后该锁也是会自动释放的。

Barrier

接着上面的应用场景讲。在推荐系统中，建完信息索引后就要开始为每个用户进行推荐了。推荐任务要分发到每台服务器上去执行，我们没有做单独的任务分发器，而是每台服务器都去同一个数据表里读取所有的用户ID，userid % n == 自己的编号时（n是服务器总数），该服务器就计算这个用户的推荐。计算推荐的过程也伴随着计算用户兴趣，所有用户的推荐计算完毕后，兴趣也就计算完毕了，此时又需要建立兴趣索引。建立兴趣索引又是只能由一台服务器来做的事情。这里有2个关键节点，即必须建完兴趣索引后所有服务器才能开始计算推荐，所有服务器计算完推荐后才能开始建兴趣索引。分布式环境下各服务器之间要想达成这种默契就需要借助于DoubleBarrier。

Barrier是指：

1）所有的线程都到达barrier后才能进行后续的计算

或者

2）所有的线程都完成自己的计算后才能离开barrier

Double Barrier是指同时具有上述两点。

Double Barrier的实现：

enter barrier:
1.建一个根节点"/root"
2.想进入barrier的线程在"/root"下建立一个子节点"/root/c_i"
3.循环监听"/root"孩子节点数目的变化，当其达到size时就说明有size个线程都已经barrier点了

leave barrier:
1.想离开barrier的线程删除其在"/root"下建立的子节点
2.循环监听"/root"孩子节点数目的变化，当size减到0时它就可以离开barrier了

服务注册

继续研究上面的应用场景，我们提到每台服务器遇到“userid % n == 自己的编号时（n是服务器总数）”这样的用户时才为其计算推荐，这里有两个问题：

集群中服务器的总数如何获取？如果直接设置成上线的服务器的个数会存在2个问题：将来服务器数目增加了n还得跟着改；如果哪天某台服务器进程挂了，那就造成1/n的用户没有推荐数据。
本服务器在集群中的编号如何获得？

解决办法是：

每台服务器进程启动时在特定的zookeeper路径下添加一个EPHEMERAL节点，节点是存储的数据为自己的IP（或者其他能唯一标识一台服务器的东西）。之所以要求是EPHEMERAL类型，是因为当进程死掉后该zookeeper节点会自动被删除掉。每天凌晨每台服务器去获取特定zookeeper路径下所有的子节点，子节点数目即为集群中服务器总数。根据IP每台服务器就可以知道自己在所有的孩子节点中排名第几。

ServerCluster.java

  1 import java.util.List;
  2 import java.util.concurrent.Executors;
  3 import java.util.concurrent.ScheduledExecutorService;
  4 import java.util.concurrent.TimeUnit;
  5 
  6 import org.apache.commons.logging.Log;
  7 import org.apache.commons.logging.LogFactory;
  8 import org.apache.curator.framework.CuratorFramework;
  9 import org.apache.curator.framework.api.CreateBuilder;
 10 import org.apache.zookeeper.CreateMode;
 11 
 12 
 13 /**
 14  * 
 15  * @Author:orisun
 16  * @Since:2016-4-7
 17  * @Version:1.0
 18  */
 19 public class ServerCluster {
 20 
 21     private static Log logger = LogFactory.getLog(ServerCluster.class);
 22     private static final String BASE_PATH = ZkClient.getInstance().getBasePath() + "/cluster";
 23     private static ScheduledExecutorService exec = Executors
 24             .newSingleThreadScheduledExecutor();
 25 
 26     /**
 27      * 向集群上报自己的存在，即把自己的IP写到特定的zk节点(EPHEMERAL节点)上去
 28      */
 29     public static void reportServer() {
 30         String selfIP = NIC.getLocalIP();
 31         CuratorFramework zkClient = ZkClient.getInstance().getZkClient();
 32         boolean exists = false;
 33         try {
 34             CreateBuilder cb = zkClient.create();
 35             if (zkClient.checkExists().forPath(BASE_PATH) == null) {
 36                 cb.creatingParentsIfNeeded().forPath(BASE_PATH,
 37                         new byte[] { 0 });
 38             }
 39             List<String> children = zkClient.getChildren().forPath(BASE_PATH);
 40             if (children != null && children.indexOf(selfIP) >= 0) {
 41                 exists = true;
 42             }
 43             if (!exists) {
 44                 // EPHEMERAL节点，进程终止时zookeeper连接断开，节点自动被删除
 45                 cb.creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
 46                         .forPath(BASE_PATH + "/" + selfIP, new byte[] { 0 });
 47                 logger.info(selfIP + " add to cluster");
 48             } else {
 49                 // 如果发现cluster上已存在该IP，则5秒后再确认一下
 50                 logger.info(selfIP + " is already in cluster");
 51                 Thread.sleep(1000 * 5);
 52                 children = zkClient.getChildren().forPath(BASE_PATH);
 53                 exists = false;
 54                 if (children != null && children.indexOf(selfIP) >= 0) {
 55                     exists = true;
 56                 }
 57                 if (!exists) {
 58                     // EPHEMERAL节点，进程终止时zookeeper连接断开，节点自动被删除
 59                     cb.creatingParentsIfNeeded()
 60                             .withMode(CreateMode.EPHEMERAL)
 61                             .forPath(BASE_PATH + "/" + selfIP, new byte[] { 0 });
 62                     logger.info(selfIP + " add to cluster");
 63                 }
 64             }
 65         } catch (Exception e) {
 66             logger.fatal("report to cluster failed", e);
 67         }
 68     }
 69 
 70     /**
 71      * 向集群上报自己的存在，即把自己的IP写到特定的zk节点(EPHEMERAL节点)上去<br>
 72      * 为防止zookeeper会话断开而造成节点被删除，每隔10分钟就去写一次
 73      */
 74     public static void report() {
 75         exec.scheduleAtFixedRate(new Runnable() {
 76             @Override
 77             public void run() {
 78                 reportServer();
 79             }
 80         }, 0, 10, TimeUnit.MINUTES);
 81     }
 82 
 83     /**
 84      * 获取集群中有多少台机器
 85      * 
 86      * @return
 87      */
 88     public static int getClusterSize() {
 89         int total = 0;
 90         List<String> children = null;
 91         try {
 92             CuratorFramework zkClient = ZkClient.getInstance().getZkClient();
 93             children = zkClient.getChildren().forPath(BASE_PATH);
 94         } catch (Exception e) {
 95             logger.error("get children of " + BASE_PATH + " failed", e);
 96         }
 97         if (children != null) {
 98             total = children.size();
 99         }
100         logger.info("cluster size is " + total);
101         return total;
102     }
103 
104     /**
105      * 获取自己在集群中的编码(从0开始)
106      * 
107      * @return
108      */
109     public static int getIndexInCluster() {
110         int index = -1;
111         CuratorFramework zkClient = ZkClient.getInstance().getZkClient();
112         try {
113             List<String> children = zkClient.getChildren().forPath(BASE_PATH);
114             String selfIP = NIC.getLocalIP();
115             index = children.indexOf(selfIP);
116         } catch (Exception e) {
117             logger.fatal("get cluster info failed", e);
118         }
119         logger.info("this server's index is " + index);
120         return index;
121     }
122 }

View Code

消息通知

经常有这样的业务场景，一台服务器执行完任务A后，需要通达其他几台服务器执行任务B。当然们可以借助于MQ框架来实现这种通知的功能，但是这种消息通常一天只发一次，为其单独创建一个消息主题未免太浪费。于是我们又想到了zookeeper，各服务器去监听同一个zookeeper节点的变化，当一台服务器执行完任务A后去把zookeeper节点的值更新当时时间，其他服务器监听到zookeeper节点有变化后，先确认一下变化为UPDATE类型，然后执行任务B，执行完后再去zookeeper节点上添加一个watcher（因为watcher只能使用一次，所以需要重复监听）。在执行任务B的过程中如果节点有变化，其他服务器是感知不到的，这样就防止了任务B在同一个进程内并行执行。

配置管理

在推荐系统中有很多配置参数，我们不想因为参数的改动而走一次上线流程（要知道参数的改动还是比较频繁的）。简单的做法是参数都写入数据库中，要改参数直接改数据库即可，程序每天去跟数据库同步一次参数。这种做法缺点很明显，就是会有一天的延迟。为了能让程序能实时感知到参数的变化，我们把参数写在zookeeper上，程序去监听zookeeper节点的变化，当有变化时从zookeeper上读取最新有值即可。考虑到zookeeper的稳定时不如MySQL，我们会在MySQL上同时存一份配置参数，通过后台修改参数时会同时修改MySQL和zookeeper。每天应用程序去跟MySQL做一次同步，zookeeper仅作为实时通知的一个工具存在。

ZkParam.java

 1 /**
 2  * 
 3  *@Author:orisun 
 4  *@Since:2016-4-7  
 5  *@Version:1.0
 6  */
 7 public class ZkParam {
 8 
 9     private double value;
10     private String path;
11     private int logicid;
12 
13     public ZkParam(double defaultValue, String path, int logicid) {
14         this.value = defaultValue;
15         this.path = path;
16         this.logicid = logicid;
17     }
18 
19     public double getValue() {
20         return value;
21     }
22 
23     public void setValue(double value) {
24         this.value = value;
25     }
26 
27     public String getPath() {
28         return path;
29     }
30 
31     public void setPath(String path) {
32         this.path = path;
33     }
34 
35     public int getLogicid() {
36         return logicid;
37     }
38 
39     public void setLogicid(int logicid) {
40         this.logicid = logicid;
41     }
42 }

View Code

ZkConfig.java

  1 import java.lang.reflect.Field;
  2 import java.lang.reflect.InvocationTargetException;
  3 import java.lang.reflect.Method;
  4 import java.util.concurrent.ExecutorService;
  5 import java.util.concurrent.Executors;
  6 
  7 import org.apache.commons.logging.Log;
  8 import org.apache.commons.logging.LogFactory;
  9 import org.apache.curator.framework.CuratorFramework;
 10 import org.apache.curator.framework.api.CreateBuilder;
 11 import org.apache.curator.framework.recipes.cache.NodeCache;
 12 import org.apache.curator.framework.recipes.cache.NodeCacheListener;
 13 
 14 /**
 15  * 
 16  * @Author:orisun
 17  * @Since:2016-4-7
 18  * @Version:1.0
 19  */
 20 public abstract class ZkConfig {
 21 
 22     private static Log logger = LogFactory.getLog(ZkConfig.class);
 23     private static final String ZK_PARAM_NAME = ZkParam.class
 24             .getCanonicalName();
 25     private Class<?> zkArgClz = null;
 26     private static Method getZkPathMethod = null;
 27     private static Method getIdMethod = null;
 28     private static Method getValueMethod = null;
 29     private ExecutorService exec = null;
 30 
 31     public ZkConfig() {
 32         exec = Executors.newCachedThreadPool();
 33         try {
 34             zkArgClz = Class.forName(ZK_PARAM_NAME);
 35             getZkPathMethod = zkArgClz.getMethod("getPath");
 36             getIdMethod = zkArgClz.getMethod("getLogicid");
 37             getValueMethod = zkArgClz.getMethod("getValue");
 38         } catch (Exception e) {
 39             logger.fatal("build " + ZK_PARAM_NAME + " failed", e);
 40             System.exit(1);
 41         }
 42     }
 43 
 44     public void updateParam(String filedName, ZkParam newArgument) {
 45         try {
 46             Method method = this.getClass().getMethod(pareSetName(filedName),
 47                     ZkParam.class);
 48             method.invoke(this, newArgument);
 49             logger.info("set " + filedName + " to " + newArgument.getValue());
 50         } catch (NoSuchMethodException | SecurityException
 51                 | IllegalAccessException | IllegalArgumentException
 52                 | InvocationTargetException e) {
 53             logger.error(
 54                     "zookeeper node is changed, but update system parameter failed",
 55                     e);
 56         }
 57     }
 58 
 59     /**
 60      * 添加zookeeper监听，参数变化时及时反应到推荐系统中
 61      */
 62     @SuppressWarnings("resource")
 63     private void addListener() {
 64         final CuratorFramework zkClient = ZkClient.getInstance().getZkClient();
 65         try {
 66             Field[] fields = this.getClass().getDeclaredFields();// 父类中的成员获取不到
 67             for (final Field field : fields) {
 68                 field.setAccessible(true);
 69                 if (field.getType().getCanonicalName().equals(ZK_PARAM_NAME)) {
 70                     Object zkParamInst = field.get(this);
 71                     final String path = (String) getZkPathMethod
 72                             .invoke(zkParamInst);
 73                     if (zkClient.checkExists().forPath(path) != null) {
 74                         final int logicid = (int) getIdMethod
 75                                 .invoke(zkParamInst);
 76                         NodeCache nodeCache = new NodeCache(zkClient, path,
 77                                 false);
 78                         nodeCache.start(true);
 79                         nodeCache.getListenable().addListener(
 80                                 new NodeCacheListener() {
 81                                     @Override
 82                                     public void nodeChanged() throws Exception {
 83                                         byte[] brr = zkClient.getData()
 84                                                 .forPath(path);
 85                                         double newValue = Double.parseDouble(new String(
 86                                                 brr));
 87                                         ZkParam newArgument = new ZkParam(
 88                                                 newValue, path, logicid);
 89                                         updateParam(field.getName(),
 90                                                 newArgument);
 91                                     }
 92                                 }, exec);
 93                         logger.info("add listener to " + path);
 94                     } else {
 95                         logger.error("will add listner on zookeeper path "
 96                                 + path + ", but it dose not exists");
 97                     }
 98                 }
 99             }
100         } catch (Exception e) {
101             logger.error("add listener to zookeeper failed", e);
102             SendMail.sendMail(SystemConfig.getValue("mail_subject"),
103                     SystemConfig.getValue("mail_receiver"),
104                     "add listener to zookeeper failed<br>" + e.getMessage());
105         }
106     }
107 
108     /**
109      * 每天定时任务，从MySQL中读取参数的值及参数对应的zkpath，然后监听该zkpath。<br>
110      * 这是为了防止zookeeper连不上，或watcher机制失效
111      */
112     public void readFromMysql() {
113         try {
114             final ParamConfigDao dao = new ParamConfigDao();
115             Field[] fields = this.getClass().getDeclaredFields();
116             for (Field field : fields) {
117                 field.setAccessible(true);
118                 if (field.getType().getCanonicalName().equals(ZK_PARAM_NAME)) {
119                     Object zkParamInst = field.get(this);
120                     String path0 = (String) getZkPathMethod.invoke(zkParamInst);
121                     int logicid = (int) getIdMethod.invoke(zkParamInst);
122                     ParamConfig param = dao.getByLogicId(logicid);
123                     if (param != null) {
124                         String path1 = param.getZkpath();
125                         String path = (path1 != null && path1.length() > 0) ? path1
126                                 : path0;
127                         if (param != null) {
128                             ZkParam newArgument = new ZkParam(param.getValue(),
129                                     path, logicid);
130                             Method method = this.getClass()
131                                     .getMethod(pareSetName(field.getName()),
132                                             ZkParam.class);
133                             method.invoke(this, newArgument);
134                         }
135                     } else {
136                         logger.error("have no such param whoese logicid is "
137                                 + logicid + " im mysql");
138                     }
139                 }
140             }
141         } catch (Exception e) {
142             logger.error("read param from mysql failed", e);
143             SendMail.sendMail(SystemConfig.getValue("mail_subject"),
144                     SystemConfig.getValue("mail_receiver"),
145                     "read param from mysql failed<br>" + e.getMessage());
146         }
147         addListener();
148     }
149 
150     /**
151      * 把参数写入到zookeeper
152      * 
153      * 
154      */
155     public void flushToZookeeper() {
156         try {
157             CuratorFramework zkClient = ZkClient.getInstance().getZkClient();
158             CreateBuilder cb = zkClient.create();
159             Field[] fields = this.getClass().getDeclaredFields();
160             for (final Field field : fields) {
161                 field.setAccessible(true);
162                 if (field.getType().getCanonicalName().equals(ZK_PARAM_NAME)) {
163                     Object zkParamInst = field.get(this);
164                     String path = (String) getZkPathMethod.invoke(zkParamInst);
165                     // 写之前先保证路径是存在的
166                     if (zkClient.checkExists().forPath(path) == null) {
167                         cb.creatingParentsIfNeeded().forPath(path,
168                                 new byte[] { 0 });
169                     }
170                     double value = (double) getValueMethod.invoke(zkParamInst);
171                     zkClient.setData().forPath(path,
172                             String.valueOf(value).getBytes());
173                 }
174             }
175         } catch (Exception e) {
176             logger.error("flush behavior weight to zookeeper failed", e);
177             SendMail.sendMail(
178                     SystemConfig.getValue("mail_subject"),
179                     SystemConfig.getValue("mail_receiver"),
180                     "flush behavior weight to zookeeper failed<br>"
181                             + e.getMessage());
182         }
183     }
184 
185     /**
186      * 拼接某属性set 方法
187      * 
188      * @param fldname
189      * @return
190      */
191     public static String pareSetName(String fldname) {
192         if (null == fldname || "".equals(fldname)) {
193             return null;
194         }
195         String pro = "set" + fldname.substring(0, 1).toUpperCase()
196                 + fldname.substring(1);
197         return pro;
198     }
199 
200     /**
201      * 拼接某属性get 方法
202      * 
203      * @param fldname
204      * @return
205      */
206     public static String pareGetName(String fldname) {
207         if (null == fldname || "".equals(fldname)) {
208             return null;
209         }
210         String pro = "get" + fldname.substring(0, 1).toUpperCase()
211                 + fldname.substring(1);
212         return pro;
213     }
214 }

View Code

使用示例

 1 /**
 2  * 
 3  * @Author:orisun
 4  * @Since:2016-4-7
 5  * @Version:1.0
 6  */
 7 public class TalentRecParamConfig extends ZkConfig {
 8 
 9     private static final String BASE_PATH = ZkClient.getInstance().getBasePath() + "/talent_rec";
10     private static final String MAX_REC_NUM_PATH = BASE_PATH + "/max_rec";
11     private static final String MIN_TITLE_SIM_PATH = BASE_PATH
12             + "/min_title_sim";
13     /** 每个职位最多推荐多少个用户 **/
14     private ZkParam maxRec = new ZkParam(1000, MAX_REC_NUM_PATH, 21);
15     /** 最小标题相似度阈值 **/
16     private ZkParam mimTitleSim = new ZkParam(0.55, MIN_TITLE_SIM_PATH, 22);// TODO
17                                                                             // 该阈值可能还需要提升
18 
19     private static volatile TalentRecParamConfig instance = null;
20 
21     private TalentRecParamConfig() {
22         super();
23     }
24 
25     /**
26      * 单例
27      * 
28      * @return
29      */
30     public static TalentRecParamConfig getInstance() {
31         if (instance == null) {
32             synchronized (TalentRecParamConfig.class) {
33                 if (instance == null) {
34                     instance = new TalentRecParamConfig();
35                 }
36             }
37         }
38         return instance;
39     }
40 
41     public ZkParam getMaxRec() {
42         return maxRec;
43     }
44 
45     public void setMaxRec(ZkParam maxRec) {
46         this.maxRec = maxRec;
47     }
48 
49     public ZkParam getMimTitleSim() {
50         return mimTitleSim;
51     }
52 
53     public void setMimTitleSim(ZkParam mimTitleSim) {
54         this.mimTitleSim = mimTitleSim;
55     }
56 
57 }