开关栅栏

    1. import java.io.Serializable;
    2. import java.util.Map;
    3. import java.util.concurrent.ConcurrentHashMap;
    4. import java.util.concurrent.locks.AbstractQueuedSynchronizer;
    6. /**
    7.  * 开关栅栏，名字跟{@link java.util.concurrent.CyclicBarrier} 有点相似，也是控制多个线程的，但是概念上不同，{@link SwitchBarrier} 是通过api的方式在代码层面上对任意个数线程进行控制，
    8.  * 而{@link java.util.concurrent.CyclicBarrier}中线程个数是指定的，而且阻塞是自动释放。在一些场景中无法使用，我们这里借鉴栅栏（Barrier）的概念，通过开和关的方式对不定个数的多线程进行群控
    9.  * {@code
    10.  * SwitchBarrier switchBarrier = new SwitchBarrier();
    11.  * switchBarrier.forbidCross();
    12.  * ...
    13.  * <p>
    14.  * // 如下执行到如下的代码的线程会阻塞掉，只有其他位置执行{@code switchBarrier.allowCross()}后，阻塞在这里的线程才会再次执行
    15.  * switchBarrier.cross();
    16.  * }
    17.  * <p>
    18.  * 这里也额外提供了一个静态内置map，也可以通过静态方式的方式对某个"开关栅栏"进行开启和关闭
    19.  * {@code
    20.  * // 一旦执行这里，则经过SwitchBarrier.cross("xxx");的线程会被阻塞
    21.  * SwitchBarrier.forbidCross("xxx");
    22.  * ...
    23.  * // 如果'栅栏'关闭，则会阻塞在这里，等待allowCross("xxx")执行后才能再次执行
    24.  * SwitchBarrier.cross("xxx");
    25.  * ...
    26.  * // 执行这里，则阻塞在 SwitchBarrier.cross("xxx"); 这里的线程会重新被唤醒
    27.  * SwitchBarrier.allowCross("xxx");
    28.  * }
    29.  * <p>
    30.  * 提示：
    31.  * 开启多次和开启一次效果是一样的，同样，关闭多次和关闭一次效果也是一样的，穿过一次和穿过多次效果也是一样（前提是开关没有变化）
    32.  *
    33.  * @author shizi
    34.  * @since 2020-11-23 12:01:36
    35.  */
    36. public class SwitchBarrier {
    38.     private final Sync sync;
    39.     /**
    40.      * 开关栅栏名称map
    41.      */
    42.     private static final Map<String, SwitchBarrier> SWITCH_BARRIER_MAP = new ConcurrentHashMap<>();
    44.     /**
    45.      * state为0认为栅栏放下（关闭），不允许通行，为1或其他则表示栅栏拉起（开启），允许通行
    46.      */
    47.     private static final class Sync extends AbstractQueuedSynchronizer implements Serializable {
    49.         Sync() {
    50.             // 可以通行
    51.             allowCross();
    52.         }
    54.         boolean canCross() {
    55.             return 0 != getState();
    56.         }
    58.         void allowCross() {
    59.             setState(1);
    60.         }
    62.         void forbidCross() {
    63.             setState(0);
    64.         }
    66.         /**
    67.          * 这里只要返回小于0 的，在share阻塞模式中就会进行阻塞
    68.          *
    69.          * @param acquires 忽略
    70.          * @return {@code -1} 在state为0时候进行阻塞使用
    71.          */
    72.         @Override
    73.         protected int tryAcquireShared(int acquires) {
    74.             return 0 == getState() ? -1 : 1;
    75.         }
    77.         /**
    78.          * 释放
    79.          *
    80.          * @param releases release个数
    81.          * @return {@code true}
    82.          */
    83.         @Override
    84.         protected boolean tryReleaseShared(int releases) {
    85.             return true;
    86.         }
    87.     }
    89.     public SwitchBarrier() {
    90.         sync = new Sync();
    91.     }
    93.     /**
    94.      * 开启栅栏，禁止通过
    95.      * <p>
    96.      * 该函数执行后，执行函数{@link SwitchBarrier#cross()}的线程就会阻塞在这里
    97.      */
    98.     public void forbidCross() {
    99.         sync.forbidCross();
    100.     }
    102.     /**
    103.      * 开启某个栅栏
    104.      * <p>
    105.      * 该函数执行后，执行函数{@link SwitchBarrier#cross(String)}的线程就会阻塞在这里
    106.      */
    107.     public static void forbidCross(String name) {
    108.         SwitchBarrier switchBarrierCache = SWITCH_BARRIER_MAP.get(name);
    109.         if (null != switchBarrierCache) {
    110.             switchBarrierCache.forbidCross();
    111.         } else {
    112.             SwitchBarrier switchBarrier = new SwitchBarrier();
    113.             switchBarrier.forbidCross();
    114.             SWITCH_BARRIER_MAP.put(name, switchBarrier);
    115.         }
    116.     }
    118.     /**
    119.      * 关闭栅栏，允许通过
    120.      * <p>
    121.      * 该函数执行后，会通知阻阻塞在函数{@link SwitchBarrier#cross()}的线程重新执行，同时后续经过函数{@link SwitchBarrier#cross()}会直接通过
    122.      */
    123.     public void allowCross() {
    124.         sync.allowCross();
    125.         sync.releaseShared(1);
    126.     }
    128.     /**
    129.      * 关闭某个栅栏
    130.      * <p>
    131.      * 该函数执行后，会通知阻塞在函数{@link SwitchBarrier#cross(String)}这里的多个线程重新执行，同时后续经过函数{@link SwitchBarrier#cross(String)}会直接通过
    132.      */
    133.     public static void allowCross(String name) {
    134.         SwitchBarrier switchBarrierCache = SWITCH_BARRIER_MAP.get(name);
    135.         if (null != switchBarrierCache) {
    136.             switchBarrierCache.allowCross();
    137.         } else {
    138.             SwitchBarrier switchBarrier = new SwitchBarrier();
    139.             switchBarrier.allowCross();
    140.             SWITCH_BARRIER_MAP.put(name, switchBarrier);
    141.         }
    142.     }
    144.     /**
    145.      * 通过栅栏
    146.      * <p>
    147.      * 如果开启了栅栏，则经过该函数的线程阻塞，否则直接通过
    148.      */
    149.     public void cross() throws InterruptedException {
    150.         sync.acquireSharedInterruptibly(0);
    151.     }
    153.     public static void cross(String name) throws InterruptedException {
    154.         SwitchBarrier switchBarrierCache = SWITCH_BARRIER_MAP.get(name);
    155.         if (null != switchBarrierCache) {
    156.             switchBarrierCache.cross();
    157.         } else {
    158.             SwitchBarrier switchBarrier = new SwitchBarrier();
    159.             switchBarrier.cross();
    160.             SWITCH_BARRIER_MAP.put(name, switchBarrier);
    161.         }
    162.     }
    164.     /**
    165.      * 栅栏是否可以通过
    166.      */
    167.     public boolean canCross() {
    168.         return sync.canCross();
    169.     }
    171.     /**
    172.      * 某个栅栏是否可以通过
    173.      */
    174.     public static boolean canCross(String name) {
    175.         SwitchBarrier switchBarrierCache = SWITCH_BARRIER_MAP.get(name);
    176.         if (null != switchBarrierCache) {
    177.             return switchBarrierCache.canCross();
    178.         } else {
    179.             SwitchBarrier switchBarrier = new SwitchBarrier();
    180.             SWITCH_BARRIER_MAP.put(name, switchBarrier);
    181.             return switchBarrier.canCross();
    182.         }
    183.     }
    185.     public static void put(String name, SwitchBarrier switchBarrier) {
    186.         SWITCH_BARRIER_MAP.put(name, switchBarrier);
    187.     }
    189.     public static SwitchBarrier getSwitchBarrier(String name) {
    190.         return SWITCH_BARRIER_MAP.get(name);
    191.     }
    192. }

    测试用例

    1. public class SwitchBarrierDemo {
    3.     ExecutorService executor = Executors.newCachedThreadPool();
    4.     private ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1, r -> {
    5.         Thread thread = new Thread(r, "pivot-client-watch-daemon-thread");
    6.         thread.setDaemon(true);
    7.         return thread;
    8.     });
    10.     SwitchBarrier switchBarrier = new SwitchBarrier();
    11.     Integer count = 7;
    12.     Integer available = 0;
    14.     @SneakyThrows
    15.     @Test
    16.     public void test1() {
    17.         scheduler.scheduleWithFixedDelay(() -> {
    18.             if (count <= 0) {
    19.                 // 在3次之后就不可用了
    20.                 switchBarrier.forbidCross();
    21.                 show("服务不可用了 ************** ");
    22.                 available++;
    23.                 if(available == 10) {
    24.                     count = 7;
    25.                     available = 0;
    26.                 }
    27.             } else {
    28.                 show("服务可用 ------------- ");
    29.                 switchBarrier.allowCross();
    30.                 count--;
    31.             }
    32.         }, 1, 1, TimeUnit.SECONDS);
    34.         for (Integer index = 0; index < 3; index++) {
    35.             Integer finalIndex = index;
    36.             executor.submit(() -> {
    37.                 while (true) {
    38.                     switchBarrier.cross();
    39.                     show("请求服务：" + finalIndex);
    40.                     // 监听应用端口，对应的应用
    41.                     Thread.sleep(2 * 1000);
    42.                     show("收到服务返回：" + finalIndex);
    44.                 }
    45.             });
    46.         }
    48.         Thread.sleep(1000 * 1000);
    49.     }
    50. }

    输出打印

    1. 请求服务：0
    2. 请求服务：1
    3. 请求服务：2
    4. 服务可用 ------------- 
    5. 收到服务返回：1
    6. 收到服务返回：2
    7. 请求服务：2
    8. 收到服务返回：0
    9. 请求服务：1
    10. 请求服务：0
    11. 服务可用 ------------- 
    12. 服务可用 ------------- 
    13. 收到服务返回：2
    14. 请求服务：2
    15. 收到服务返回：1
    16. 请求服务：1
    17. 收到服务返回：0
    18. 请求服务：0
    19. 服务可用 ------------- 
    20. 服务可用 ------------- 
    21. 收到服务返回：0
    22. 请求服务：0
    23. 收到服务返回：1
    24. 请求服务：1
    25. 收到服务返回：2
    26. 请求服务：2
    27. 服务可用 ------------- 
    28. 服务可用 ------------- 
    29. 收到服务返回：1
    30. 收到服务返回：0
    31. 收到服务返回：2
    32. 请求服务：0
    33. 请求服务：1
    34. 请求服务：2
    35. 服务不可用了 ************** 
    36. 服务不可用了 ************** 
    37. 收到服务返回：2
    38. 收到服务返回：1
    39. 收到服务返回：0
    40. 服务不可用了 ************** 
    41. 服务不可用了 ************** 
    42. 服务不可用了 ************** 
    43. 服务不可用了 ************** 
    44. 服务不可用了 ************** 
    45. 服务不可用了 ************** 
    46. 服务不可用了 ************** 
    47. 服务不可用了 ************** 
    48. 服务可用 ------------- 
    49. 请求服务：2
    50. 请求服务：1
    51. 请求服务：0
    52. 服务可用 ------------- 
    53. 收到服务返回：0
    54. 请求服务：0
    55. 收到服务返回：1
    56. 请求服务：1
    57. 收到服务返回：2
    58. 请求服务：2
    59. 服务可用 ------------- 
    60. 服务可用 ------------- 
    61. 收到服务返回：0
    62. 请求服务：0
    63. 收到服务返回：2
    64. 请求服务：2
    65. 收到服务返回：1
    66. 请求服务：1
    67. 服务可用 ------------- 
    68. 服务可用 ------------- 
    69. 收到服务返回：0
    70. 请求服务：0
    71. 收到服务返回：2
    72. 请求服务：2
    73. 收到服务返回：1
    74. 请求服务：1
    75. 服务可用 ------------- 
    76. 服务不可用了 ************** 
    77. 收到服务返回：2
    78. 收到服务返回：0
    79. 收到服务返回：1
    80. 服务不可用了 ************** 
    81. 服务不可用了 ************** 
    82. 服务不可用了 ************** 
    83. 服务不可用了 ************** 
    84. 服务不可用了 ************** 
    85. 服务不可用了 ************** 
    86. 服务不可用了 ************** 
    87. 服务不可用了 ************** 
    88. 服务不可用了 ************** 
    89. 服务可用 ------------- 
    90. 请求服务：2
    91. 请求服务：1
    92. 请求服务：0
    93. 服务可用 ------------- 
    94. 收到服务返回：1
    95. 收到服务返回：0
    96. 请求服务：0
    97. 收到服务返回：2
    98. 请求服务：2
    99. 请求服务：1
    100. 服务可用 ------------- 
    101. 服务可用 -------------