环境是这æ ïLš„åQšæœåŠ¡å™¨æ˜¯ç”¨Java做的åQ?数据库是MongoDB

需求是˜q™æ ·çš„：我们的系¾lŸé‡Œè¦ç”Ÿæˆä¸€ä¸ªå”¯ä¸€IDåQŒå‰é¢çš„部分有一定的格式åQŒåƈ和时间关联， ¾_„¡¡®åˆ°å¾®¿U’，考虑到同一微秒内有可能存在òq¶å‘情况åQ? 所以后面在加两位序列号åQ?¾pȝ»Ÿéœ€è¦å®šä¹‰äØ“1毫秒内的òq¶å‘ž®äºŽ100个，所以后面两位就够用了ã€? Java服务器端有多台机器都可以用来生成˜q™ä¸ªå”¯ä¸€IDåQŒæ‰€ä»¥éœ€è¦åœ¨ä¸åŒçš„机器上不能生成相同的序列号åQŒæ‰€ä»¥éœ€è¦åœ¨æŸä¸€ç‚¹ä¸Šåšå…¨å±€çš„范围同步来保存˜q™åºåˆ? åïLš„唯一性ã€?其实如果不考虑需求里的唯一ID是有一定意义的格式的， 用UUID或MongoDBçš„ObjectId都是更好的选择åQŒå®Œå…¨ä¸éœ€è¦åœ¨æŸä¸€ç‚¹ä¸Š˜q›è¡ŒåŒæ­¥åQŒæ€§èƒ½ä¼šæ›´å¥½ã€?/p>

˜q™ä¸ªå¯ä»¥ç”Ÿæˆåºåˆ—åïLš„点， 我们可以做一个序列号生成服务器来对应åQ?也可以用数据库来对应ã€? 单单ä¸ø™¿™ä¸ªç®€å•çš„功能准备一个服务器来做昄¡„¶ä¸åˆé€‚ã€?但是我们用的MongoDBòq¶æ²¡æœ‰ç±»ä¼égºŽMySQL或Oracle中的SELECT FOR UPDATE˜q™æ ·çš„锁机制ã€?所以没有办法简单的对这个序列号做原子操作ã€? 但是MongoDB的对单个document˜q›è¡Œupdate操作中有很是å…ähœ‰åŽŸå­æ€§çš„åQ?例如

• $set
• $unset
• $inc
• $push
• $pushAll
• $pull
• $pullAll

我们可以利用˜q™äº›åŽŸå­æ“ä½œåQŒåœ¨æ•°æ®åº“层以乐观锁的åŞ式来实现循环序列字段。äؓ了方便调用我把这ŒDµé€»è¾‘做成数据库中的Javascript函数ã€?¾cÖM¼¼ä¸ŽMySQL中的存储˜q‡ç¨‹ã€?/p>

首先我们需要一个collection来存攑ֺåˆ—号åQŒåƈ寚wœ€è¦çš„需要的序列可‚¿›è¡Œåˆå§‹åŒ–。我们叫它countersã€?/p>

Js代码  

1. db.counters.save({_id:"SerialNo1", val:0, maxval:99})  

然后我们想system.js里添加一个Javascript函数

Js代码  

1. db.system.js.save({_id:"getNextUniqueSeq",  
2. value:function (keyName) {  
3.     var seqObj = db.counters.findOne({_id:keyName});  
4.     if (seqObj == null) {  
5.         print("can not find record with key: " + keyName);  
6.         return -1;  
7.     }  
8.       
9.     // the max value of sequence  
10.     var maxVal = seqObj.maxval;  
11.     // the current value of sequence  
12.     var curVal = seqObj.val;  
13.       
14.     while(true){  
15.         // if curVal reach max, reset it  
16.         if(curVal >= maxVal){  
17.             db.counters.update({_id : keyName, val : curVal}, { $set : { val : 0 }}, false, false);  
18.             var err = db.getLastErrorObj();  
19.             if( err && err.code ) {  
20.                 print( "unexpected error reset data: " + tojson( err ) );  
21.                 return -2;  
22.             } else if (err.n == 0){  
23.                 // fail to reset value, may be reseted by others  
24.                 print("fail to reset value: ");  
25.             }   
26.   
27.             // get current value again.  
28.             seqObj = db.counters.findOne({_id:keyName});  
29.             maxVal = seqObj.maxval;  
30.             curVal = seqObj.val;  
31.             continue;  
32.         }   
33.           
34.         // if curVal not reach the max, inc it;  
35.         // increase   
36.         db.counters.update({_id : keyName, val : curVal}, { $inc : { val : 1 }}, false, false);  
37.         var err = db.getLastErrorObj();  
38.         if( err && err.code ) {  
39.             print( "unexpected error inc val: " + tojson( err ) );  
40.                return -3;  
41.         } else if (err.n == 0){  
42.             // fail to reset value, may be increased by others  
43.             print("fail to inc value: ");  
44.               
45.             // get current value again.  
46.             seqObj = db.counters.findOne({_id:keyName});  
47.             maxVal = seqObj.maxval;  
48.             curVal = seqObj.val;  
49.             continue;  
50.         } else {  
51.             var retVal = curVal + 1;  
52.             print("success to get seq : " + retVal);  
53.             // increase successful  
54.             return retVal;  
55.         }  
56.     }  
57. }  
58. });  

上面˜q™æ®µä¼šæŠŠæŒ‡å®šçš„序列号的valå€?1åQŒå¦‚æžœval辑ֈ°ä¸Šé™åˆ™ä»Ž0开始。所以叫循环序列ã€?/p>

其实上面的实现在原理上和Java里的AtomicInteger¾pÕdˆ—的功能实现是¾cÖM¼¼çš„，利用循环重试和原子性的CAS来实现。这¿Uå®žçŽ°æ–¹å¼åœ¨å¤šçº¿½E‹çš„环境里由于锁åQˆMonitoråQ‰çš„范围很小åQŒæ‰€ä»¥åƈ发性上比排他锁要好一些ã€?/p>

下面我们用Java来测试一下这个函数的正确性ã€?卛_œ¨å¤šçº¿½E‹çš„情况下会不会得到重复的序列号ã€?/p>

½W¬ä¸€ä¸ªæµ‹è¯•ï¼Œval=0åQ?maxval=2000åQ?Javaç«?0个线½E‹æ¯ä¸ªçº¿½E‹åó@环调ç”?00‹Æ¡ã€?å…?000‹Æ¡ã€?所以正¼‹®çš„情况下，ä»?åˆ?999应该每个数字只出çŽîC¸€‹Æ¡ã€?/p>

Java代码  

1. @Test  
2. public void testGetNextUniqueSeq1() throws Exception {  
3.   
4.     final int THREAD_COUNT = 20;  
5.     final int LOOP_COUNT = 100;  
6.   
7.     Mongo mongoClient = new Mongo("172.17.2.100", 27017);  
8.     DB db = mongoClient.getDB("im");  
9.     db.authenticate("imadmin", "imadmin".toCharArray());  
10.     BasicDBObject q = new BasicDBObject();  
11.     q.put("_id", "UNIQUE_KEY");  
12.   
13.     BasicDBObject upd = new BasicDBObject();  
14.     BasicDBObject set = new BasicDBObject();  
15.     set.put("val", 0);  
16.     set.put("maxval", THREAD_COUNT * LOOP_COUNT);  
17.     upd.put("$set", set);  
18.   
19.     db.getCollection("counters").update(q, upd);  
20.   
21.     Thread[] threads = new Thread[THREAD_COUNT];  
22.     final int[][] results = new int[THREAD_COUNT][LOOP_COUNT];  
23.     for (int i = 0; i < THREAD_COUNT; i++) {  
24.         final int temp_i = i;  
25.         threads[i] = new Thread("" + i) {  
26.             @Override  
27.             public void run() {  
28.                 try {  
29.                     Mongo mongoClient = new Mongo("172.17.2.100", 27017);  
30.                     DB db = mongoClient.getDB("im");  
31.                     db.authenticate("imadmin", "imadmin".toCharArray());  
32.                     for (int j = 0; j < LOOP_COUNT; j++) {  
33.                         Object result = db.eval("getNextUniqueSeq(\"UNIQUE_KEY\")");  
34.                         System.out.printf("Thread %s, seq=%d\n", Thread.currentThread().getName(), ((Double) result).intValue());  
35.                         results[temp_i][j] = ((Double) result).intValue();  
36.                     }  
37.                 } catch (UnknownHostException e) {  
38.                     e.printStackTrace();  
39.                 }  
40.             }  
41.         };  
42.     }  
43.   
44.     for (Thread thread : threads) {  
45.         thread.start();  
46.     }  
47.   
48.     for (Thread thread : threads) {  
49.         thread.join();  
50.     }  
51.   
52.     for (int num = 1; num <= LOOP_COUNT * THREAD_COUNT; num++) {  
53.         // every number appear 1 times only!  
54.         int times = 0;  
55.         for (int j = 0; j < THREAD_COUNT; j++) {  
56.             for (int k = 0; k < LOOP_COUNT; k++) {  
57.                 if (results[j][k] == num)  
58.                     times++;  
59.             }  
60.         }  
61.   
62.         assertEquals(1, times);  
63.     }  
64. }  

然后我们再测试一下åó@环的情况ã€?val=0, maxval=99ã€?同样是Javaç«?0个线½E‹æ¯ä¸ªçº¿½E‹åó@环调ç”?00‹Æ¡ã€?å…?000‹Æ¡ã€‚è¿™‹Æ¡ä»Ž0åˆ?9的数字每个应该取å¾?0‹Æ¡ã€?/p>

Java代码  

1. @Test  
2. public void testGetNextUniqueSeq2() throws Exception {  
3.   
4.     final int THREAD_COUNT = 20;  
5.     final int LOOP_COUNT = 100;  
6.   
7.     Mongo mongoClient = new Mongo("172.17.2.100", 27017);  
8.     DB db = mongoClient.getDB("im");  
9.     db.authenticate("imadmin", "imadmin".toCharArray());  
10.     BasicDBObject q = new BasicDBObject();  
11.     q.put("_id", "UNIQUE_KEY");  
12.   
13.     BasicDBObject upd = new BasicDBObject();  
14.     BasicDBObject set = new BasicDBObject();  
15.     set.put("val", 0);  
16.     set.put("maxval", LOOP_COUNT);  
17.     upd.put("$set", set);  
18.   
19.     db.getCollection("counters").update(q, upd);  
20.   
21.     Thread[] threads = new Thread[THREAD_COUNT];  
22.     final int[][] results = new int[THREAD_COUNT][LOOP_COUNT];  
23.     for (int i = 0; i < THREAD_COUNT; i++) {  
24.         final int temp_i = i;  
25.         threads[i] = new Thread("" + i) {  
26.             @Override  
27.             public void run() {  
28.                 try {  
29.                     Mongo mongoClient = new Mongo("172.17.2.100", 27017);  
30.                     DB db = mongoClient.getDB("im");  
31.                     db.authenticate("imadmin", "imadmin".toCharArray());  
32.                     for (int j = 0; j < LOOP_COUNT; j++) {  
33.                         Object result = db.eval("getNextUniqueSeq(\"UNIQUE_KEY\")");  
34.                         System.out.printf("Thread %s, seq=%d\n", Thread.currentThread().getName(), ((Double) result).intValue());  
35.                         results[temp_i][j] = ((Double) result).intValue();  
36.                     }  
37.                 } catch (UnknownHostException e) {  
38.                     e.printStackTrace();  
39.                 }  
40.             }  
41.         };  
42.     }  
43.   
44.     for (Thread thread : threads) {  
45.         thread.start();  
46.     }  
47.   
48.     for (Thread thread : threads) {  
49.         thread.join();  
50.     }  
51.   
52.     for (int num = 1; num <= LOOP_COUNT; num++) {  
53.         // every number appear 20 times only!  
54.         int times = 0;  
55.         for (int j = 0; j < THREAD_COUNT; j++) {  
56.             for (int k = 0; k < LOOP_COUNT; k++) {  
57.                 if (results[j][k] == num)  
58.                     times++;  
59.             }  
60.         }  
61.   
62.         assertEquals(20, times);  
63.     }  
64. }  

˜q™ä¸ª‹¹‹è¯•è·‘了几次都是正确的ã€?/p>

ç”׃ºŽæ²¡æœ‰å¯ä»¥˜q›è¡Œå¯ÒŽ¯”其他的实现方式（例如排他锁）所以没有做性能‹¹‹è¯•ã€?/p>

写在最后ã€?虽然MongoDB支持¾cÖM¼¼äºŽå­˜å‚¨è¿‡½E‹çš„Stored JavascriptåQŒä½†æ˜¯å…¶å®žä¸å»ø™®®ä½¿ç”¨˜q™ä¸ªæ¥è§£å†›_¤æ‚问题。主要原因是没法调试åQŒç»´æŠ¤è“v来太不方ä¾Ñ€‚而且åœ?.4之前MongoDBå¯ÒŽœåŠ¡ç«¯ Javascript支持òq¶ä¸æ˜¯å¾ˆå¥½ï¼Œ 一个mongod˜q›ç¨‹åŒæ—¶åªèƒ½æ‰§è¡Œä¸€ŒDµJavascript。如果能在应用层解决掉还是在应用层里实现逻辑比较好ã€?/p>