非公平訊號量

說明

1.通過zset add 和rank來實現是否獲取訊號量的判斷,

2.add時通過當前時間+超時時間 計算的時間設定為score 每次add提前刪除過期的0~當前時間

訊號量類封裝

    public static class RedisSemaphore {
        //執行緒快取儲存index 用於釋放
        ThreadLocal<String> semaphoreValue = new ThreadLocal<>();
        private Integer limit;

        public RedisSemaphore( Integer limit) {
            
 
//因為redis rank從0開始 所以 limit-1
            this.limit = limit-1;
        }

        /**
         * 訊號量
         * @param timeout
         * @return
         */
        public boolean acquire( Jedis conn,Integer timeout) {
            String index = UUID.randomUUID().toString();
            //計算過期時間
            Calendar c = Calendar.getInstance();
            c.setTime(
 
new Date());
            c.add(Calendar.SECOND, timeout);
            conn.zadd("semaphore:acquire", c.getTime().getTime(), index);
            //刪除過期的
            conn.zremrangeByScore("semaphore:acquire", 0, System.currentTimeMillis());
            //判斷是否獲得訊號量 根據獲得的排名來
            Long rank = conn.zrank("semaphore:acquire", index);
            
 
if (rank > limit) {
                //刪除
                conn.zrem("semaphore:acquire", index);
                return false;
            }
            //執行緒快取儲存用於釋放
            semaphoreValue.set(index);
            return true;

        }

        public void release( Jedis conn) {
            String index = semaphoreValue.get();
            semaphoreValue.remove();;
            if (index == null) {
                return;
            }
            conn.zrem("semaphore:acquire", index);

        }
    }

測試類

public static void main(String[] args)
            throws Exception {
        Jedis conn = new Jedis("127.0.0.1", 6379);
        conn.flushDB();
        RedisSemaphore redisSemaphore=new RedisSemaphore(11);
        CountDownLatch countDownLatch=new CountDownLatch(11);
        //=====================獲取並釋放===================
        System.out.println("==================多執行緒獲取並釋放訊號量結果==================");
        for(int i=0;i<11;i++){
            final int x=i;
            new Thread(new Runnable() {
                int j=x;
                @Override
                public void run() {
                    Jedis conn = new Jedis("127.0.0.1", 6379);

                    System.out.println("獲取釋放,i"+j+","+redisSemaphore.acquire(conn,30));
                    redisSemaphore.release(conn);
                    countDownLatch.countDown();
                }
            }).start();

        }
        countDownLatch.await();
        //=====================不釋放===================
        System.out.println("==================多執行緒不釋放獲取訊號量結果 下面正常獲取表示有上面有正常釋放==================");

        for(int i=0;i<12;i++){
            final int x=i;
            new Thread(new Runnable() {
                int j=x;
                @Override
                public void run() {
                    Jedis conn = new Jedis("127.0.0.1", 6379);

                    System.out.println("獲取不釋放i"+j+","+redisSemaphore.acquire(conn,30));

                }
            }).start();

        }

    }

列印

==================多執行緒獲取並釋放訊號量結果==================
獲取釋放,i8,true
獲取釋放,i7,true
獲取釋放,i1,true
獲取釋放,i10,true
獲取釋放,i3,true
獲取釋放,i6,true
獲取釋放,i0,true
獲取釋放,i4,true
獲取釋放,i2,true
獲取釋放,i9,true
獲取釋放,i5,true
==================多執行緒不釋放獲取訊號量結果 下面正常獲取表示有上面有正常釋放==================
獲取不釋放i1,true
獲取不釋放i0,true
獲取不釋放i2,true
獲取不釋放i3,true
獲取不釋放i4,true
獲取不釋放i5,true
獲取不釋放i8,true
獲取不釋放i7,true
獲取不釋放i6,false
獲取不釋放i11,true
獲取不釋放i10,true
獲取不釋放i9,true