互聯網面試的時候，是不是面試官常問一個問題如何保證集群環境下數據操作并發問題，常用的synchronized肯定是無法滿足了，或許你可以借助for update對數據加鎖。本文的最終解決方式你只要在方法上加一個@P4jSyn注解就能保證集群環境下同synchronized的效果，且鎖的key可以任意指定。本注解還支持了鎖的超時機制。
本文需要對Redis、spring和spring-data-redis有一定的了解。當然你可以借助本文的思路對通過注解對方法返回數據進行緩存，類似com.google.code.simple-spring-memcached的@ReadThroughSingleCache。
第一步： 介紹兩個自定義注解P4jSyn、P4jSynKey
P4jSyn：必選項，標記在方法上，表示需要對該方法加集群同步鎖；
P4jSynKey：可選項，加在方法參數上，表示以方法某個參數作為鎖的key，用來保證更多的坑，P4jSynKey并不是強制要添加的，當沒有P4jSynKey標記的情況下只會以P4jSyn的synKey作為鎖key。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
package com.yaoguoyin.redis.lock; import java.lang.annotation.ElementType; import java.lang.annotation.Inherited; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; /** * <b>同步鎖：</b><br/> * 主要作用是在服務器集群環境下保證方法的synchronize；<br/> * 標記在方法上，使該方法的執行具有互斥性，并不保證并發執行方法的先后順序；<br/> * 如果原有“A任務”獲取鎖后任務執行時間超過最大允許持鎖時間，且鎖被“B任務”獲取到，在“B任務”成功貨物鎖會并不會終止“A任務”的執行；<br/> * <br/> * <b>注意：</b><br/> * 使用過程中需要注意keepMills、toWait、sleepMills、maxSleepMills等參數的場景使用；<br/> * 需要安裝redis，并使用spring和spring-data-redis等，借助redis NX等方法實現。 * * @see com.yaoguoyin.redis.lock.P4jSynKey * @see com.yaoguoyin.redis.lock.RedisLockAspect * * @author partner4java * */@Target({ ElementType.METHOD }) @Retention(RetentionPolicy.RUNTIME) @Inheritedpublic @interface P4jSyn { /** * 鎖的key<br/> * 如果想增加坑的個數添加非固定鎖，可以在參數上添加@P4jSynKey注解，但是本參數是必寫選項<br/> * redis key的拼寫規則為 "RedisSyn+" + synKey + @P4jSynKey<br/> * */ String synKey(); /** * 持鎖時間，超時時間，持鎖超過此時間自動丟棄鎖<br/> * 單位毫秒,默認20秒<br/> * 如果為0表示永遠不釋放鎖，在設置為0的情況下toWait為true是沒有意義的<br/> * 但是沒有比較強的業務要求下，不建議設置為0 */ long keepMills() default 20 * 1000; /** * 當獲取鎖失敗，是繼續等待還是放棄<br/> * 默認為繼續等待 */ boolean toWait() default true; /** * 沒有獲取到鎖的情況下且toWait()為繼續等待，睡眠指定毫秒數繼續獲取鎖，也就是輪訓獲取鎖的時間<br/> * 默認為10毫秒 * * @return */ long sleepMills() default 10; /** * 鎖獲取超時時間：<br/> * 沒有獲取到鎖的情況下且toWait()為true繼續等待，最大等待時間，如果超時拋出 * {@link java.util.concurrent.TimeoutException.TimeoutException} * ，可捕獲此異常做相應業務處理；<br/> * 單位毫秒,默認一分鐘，如果設置為0即為沒有超時時間，一直獲取下去； * * @return */ long maxSleepMills() default 60 * 1000; } |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
package com.yaoguoyin.redis.lock; import java.lang.annotation.ElementType; import java.lang.annotation.Inherited; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; /** * <b>同步鎖 key</b><br/> * 加在方法的參數上，指定的參數會作為鎖的key的一部分 * * @author partner4java * */@Target({ ElementType.PARAMETER }) @Retention(RetentionPolicy.RUNTIME) @Inheritedpublic @interface P4jSynKey { /** * key的拼接順序 * * @return */ int index() default 0; } |
這里就不再對兩個注解進行使用上的解釋了，因為注釋已經說明的很詳細了。
使用示例：
|
1
2
3
4
5
6
7
8
9
10
11
|
package com.yaoguoyin.redis.lock; import org.springframework.stereotype.Component; @Componentpublic class SysTest { private static int i = 0; @P4jSyn(synKey = "12345") public void add(@P4jSynKey(index = 1) String key, @P4jSynKey(index = 0) int key1) { i++; System.out.println("i=-===========" + i); } } |
第二步：切面編程
在不影響原有代碼的前提下，保證執行同步，目前最直接的方式就是使用切面編程
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
package com.yaoguoyin.redis.lock; import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.SortedMap; import java.util.TreeMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Aspect; import org.aspectj.lang.reflect.MethodSignature; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.data.redis.core.BoundValueOperations; import org.springframework.data.redis.core.RedisTemplate; /** * 鎖的切面編程<br/> * 針對添加@RedisLock 注解的方法進行加鎖 * * @see com.yaoguoyin.redis.lock.P4jSyn * * @author partner4java * */@Aspectpublic class RedisLockAspect { @Autowired @Qualifier("redisTemplate") private RedisTemplate<String, Long> redisTemplate; @Around("execution(* com.yaoguoyin..*(..)) && @annotation(com.yaoguoyin.redis.lock.P4jSyn)") public Object lock(ProceedingJoinPoint pjp) throws Throwable { P4jSyn lockInfo = getLockInfo(pjp); if (lockInfo == null) { throw new IllegalArgumentException("配置參數錯誤"); } String synKey = getSynKey(pjp, lockInfo.synKey()); if (synKey == null || "".equals(synKey)) { throw new IllegalArgumentException("配置參數synKey錯誤"); } boolean lock = false; Object obj = null; try { // 超時時間 long maxSleepMills = System.currentTimeMillis() + lockInfo.maxSleepMills(); while (!lock) { long keepMills = System.currentTimeMillis() + lockInfo.keepMills(); lock = setIfAbsent(synKey, keepMills); // 得到鎖，沒有人加過相同的鎖 if (lock) { obj = pjp.proceed(); } // 鎖設置了沒有超時時間 else if (lockInfo.keepMills() <= 0) { // 繼續等待獲取鎖 if (lockInfo.toWait()) { // 如果超過最大等待時間拋出異常 if (lockInfo.maxSleepMills() > 0 && System.currentTimeMillis() > maxSleepMills) { throw new TimeoutException("獲取鎖資源等待超時"); } TimeUnit.MILLISECONDS.sleep(lockInfo.sleepMills()); } else { break; } } // 已過期，并且getAndSet后舊的時間戳依然是過期的，可以認為獲取到了鎖 else if (System.currentTimeMillis() > getLock(synKey) && (System.currentTimeMillis() > getSet(synKey, keepMills))) { lock = true; obj = pjp.proceed(); } // 沒有得到任何鎖 else { // 繼續等待獲取鎖 if (lockInfo.toWait()) { // 如果超過最大等待時間拋出異常 if (lockInfo.maxSleepMills() > 0 && System.currentTimeMillis() > maxSleepMills) { throw new TimeoutException("獲取鎖資源等待超時"); } TimeUnit.MILLISECONDS.sleep(lockInfo.sleepMills()); } // 放棄等待 else { break; } } } } catch (Exception e) { e.printStackTrace(); throw e; } finally { // 如果獲取到了鎖，釋放鎖 if (lock) { releaseLock(synKey); } } return obj; } /** * 獲取包括方法參數上的key<br/> * redis key的拼寫規則為 "RedisSyn+" + synKey + @P4jSynKey * */ private String getSynKey(ProceedingJoinPoint pjp, String synKey) { try { synKey = "RedisSyn+" + synKey; Object[] args = pjp.getArgs(); if (args != null && args.length > 0) { MethodSignature methodSignature = (MethodSignature) pjp.getSignature(); Annotation[][] paramAnnotationArrays = methodSignature.getMethod().getParameterAnnotations(); SortedMap<Integer, String> keys = new TreeMap<Integer, String>(); for (int ix = 0; ix < paramAnnotationArrays.length; ix++) { P4jSynKey p4jSynKey = getAnnotation(P4jSynKey.class, paramAnnotationArrays[ix]); if (p4jSynKey != null) { Object arg = args[ix]; if (arg != null) { keys.put(p4jSynKey.index(), arg.toString()); } } } if (keys != null && keys.size() > 0) { for (String key : keys.values()) { synKey = synKey + key; } } } return synKey; } catch (Exception e) { e.printStackTrace(); } return null; } @SuppressWarnings("unchecked") private static <T extends Annotation> T getAnnotation(final Class<T> annotationClass, final Annotation[] annotations) { if (annotations != null && annotations.length > 0) { for (final Annotation annotation : annotations) { if (annotationClass.equals(annotation.annotationType())) { return (T) annotation; } } } return null; } /** * 獲取RedisLock注解信息 */ private P4jSyn getLockInfo(ProceedingJoinPoint pjp) { try { MethodSignature methodSignature = (MethodSignature) pjp.getSignature(); Method method = methodSignature.getMethod(); P4jSyn lockInfo = method.getAnnotation(P4jSyn.class); return lockInfo; } catch (Exception e) { e.printStackTrace(); } return null; } public BoundValueOperations<String, Long> getOperations(String key) { return redisTemplate.boundValueOps(key); } /** * Set {@code value} for {@code key}, only if {@code key} does not exist. * <p> * See http://redis.io/commands/setnx * * @param key * must not be {@literal null}. * @param value * must not be {@literal null}. * @return */ public boolean setIfAbsent(String key, Long value) { return getOperations(key).setIfAbsent(value); } public long getLock(String key) { Long time = getOperations(key).get(); if (time == null) { return 0; } return time; } public long getSet(String key, Long value) { Long time = getOperations(key).getAndSet(value); if (time == null) { return 0; } return time; } public void releaseLock(String key) { redisTemplate.delete(key); } } |
RedisLockAspect會對添加注解的方法進行特殊處理，具體可看lock方法。
大致思路就是：
1、首選借助redis本身支持對應的setIfAbsent方法，該方法的特點是如果redis中已有該數據不保存返回false，不存該數據保存返回true；
2、如果setIfAbsent返回true標識拿到同步鎖，可進行操作，操作后并釋放鎖；
3、如果沒有通過setIfAbsent拿到數據，判斷是否對鎖設置了超時機制，沒有設置判斷是否需要繼續等待；
4、判斷是否鎖已經過期，需要對(System.currentTimeMillis() > getLock(synKey) && (System.currentTimeMillis() > getSet(synKey, keepMills)))進行細細的揣摩一下，getSet可能會改變了其他人擁有鎖的超時時間，但是幾乎可以忽略；
5、沒有得到任何鎖，判斷繼續等待還是退出。
第三步：spring的基本配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
#*****************jedis連接參數設置*********************# #redis服務器ip # redis.hostName=127.0.0.1 #redis服務器端口號# redis.port=6379 #redis服務器外部訪問密碼 redis.password=XXXXXXXXXX #************************jedis池參數設置*******************# #jedis的最大分配對象# jedis.pool.maxActive=1000 jedis.pool.minIdle=100 #jedis最大保存idel狀態對象數 # jedis.pool.maxIdle=1000 #jedis池沒有對象返回時，最大等待時間 # jedis.pool.maxWait=5000 #jedis調用borrowObject方法時，是否進行有效檢查# jedis.pool.testOnBorrow=true #jedis調用returnObject方法時，是否進行有效檢查 # jedis.pool.testOnReturn=true |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p" xmlns:context="http://www.springframework.org/schema/context" xmlns:jee="http://www.springframework.org/schema/jee"xmlns:tx="http://www.springframework.org/schema/tx" xmlns:aop="http://www.springframework.org/schema/aop"xmlns:redis="http://www.springframework.org/schema/redis" xmlns:cache="http://www.springframework.org/schema/cache" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.1.xsd http://www.springframework.org/schema/redis http://www.springframework.org/schema/redis/spring-redis.xsd http://www.springframework.org/schema/cache http://www.springframework.org/schema/cache/spring-cache.xsd"> <!-- 開啟注解 --> <aop:aspectj-autoproxy /> <bean class="com.yaoguoyin.redis.lock.RedisLockAspect" /> <!-- 掃描注解包范圍 --> <context:component-scan base-package="com.yaoguoyin" /> <!-- 引入redis配置 --> <context:property-placeholder location="classpath:config.properties" /> <!-- 連接池 --> <bean id="poolConfig" class="redis.clients.jedis.JedisPoolConfig"> <property name="minIdle" value="${jedis.pool.minIdle}" /> <property name="maxIdle" value="${jedis.pool.maxIdle}" /> <property name="maxWaitMillis" value="${jedis.pool.maxWait}" /> </bean> <!-- p:password="${redis.pass}" --> <bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory" p:host-name="${redis.hostName}" p:port="${redis.port}" p:password="${redis.password}" p:pool-config-ref="poolConfig" /> <!-- 類似于jdbcTemplate --> <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate" p:connection-factory-ref="redisConnectionFactory" /> </beans> |
redis的安裝本文就不再說明。
測試
|
1
2
3
4
5
6
7
8
9
|
package com.yaoguoyin.redis; import org.junit.runner.RunWith; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = { "classpath:META-INF/spring/redis.xml" }) public class BaseTest extends AbstractJUnit4SpringContextTests { } |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
package com.yaoguoyin.redis.lock; import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; import com.yaoguoyin.redis.BaseTest; public class RedisTest extends BaseTest { @Autowired private SysTest sysTest; @Test public void testHello() throws InterruptedException { for (int i = 0; i < 100; i++) { new Thread(new Runnable() { @Override public void run() { try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } sysTest.add("xxxxx", 111111); } }).start(); } TimeUnit.SECONDS.sleep(20); } @Test public void testHello2() throws InterruptedException{ sysTest.add("xxxxx", 111111); TimeUnit.SECONDS.sleep(10); } } |
你可以對
void com.yaoguoyin.redis.lock.SysTest.add(@P4jSynKey(index=1) String key, @P4jSynKey(index=0) int key1)
去除注解@P4jSyn進行測試對比。
ps：本demo的執行性能取決于redis和Java交互距離；成千山萬單鎖并發建議不要使用這種形式，直接通過redis等解決，本demo只解決小并發不想耦合代碼的形式。
以上就是本文的全部內容，希望本文的內容對大家的學習或者工作能帶來一定的幫助，同時也希望多多支持服務器之家！
原文鏈接：http://blog.csdn.net/partner4java/article/details/52198801
