个人中心
gpt4 book ai didi

Java并发编程Semaphore计数信号量详解

转载 作者：qq735679552 更新时间：2022-09-28 22:32:09 25 4
gpt4 key购买 nike

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章Java并发编程Semaphore计数信号量详解由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

Semaphore 是一个计数信号量，它的本质是一个共享锁。信号量维护了一个信号量许可集。线程可以通过调用acquire()来获取信号量的许可；当信号量中有可用的许可时，线程能获取该许可；否则线程必须等待，直到有可用的许可为止。 线程可以通过release()来释放它所持有的信号量许可（用完信号量之后必须释放，不然其他线程可能会无法获取信号量）.

简单示例:

?
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
package me.socketthread;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class SemaphoreLearn {
   //信号量总数
   private static final int SEM_MAX = 12 ;
   public static void main(String[] args) { 
     Semaphore sem = new Semaphore(SEM_MAX);
     //创建线程池
     ExecutorService threadPool = Executors.newFixedThreadPool( 3 );
     //在线程池中执行任务
     threadPool.execute( new MyThread(sem, 7 ));
     threadPool.execute( new MyThread(sem, 4 ));
     threadPool.execute( new MyThread(sem, 2 ));
     //关闭池
     threadPool.shutdown();
   }
}
   class MyThread extends Thread {
     private volatile Semaphore sem;  // 信号量
     private int count;    // 申请信号量的大小 
     
     MyThread(Semaphore sem, int count) {
       this .sem = sem;
       this .count = count;
     }
     public void run() {
       try {
        // 从信号量中获取count个许可
         sem.acquire(count);
         Thread.sleep( 2000 );
         System.out.println(Thread.currentThread().getName() + " acquire count=" +count);
       } catch (InterruptedException e) {
         e.printStackTrace();
       } finally {
         // 释放给定数目的许可，将其返回到信号量。
         sem.release(count);
         System.out.println(Thread.currentThread().getName() + " release " + count + "" );
       }
     }
   }

执行结果:

?
1
2
3
4
5
6
pool-1-thread-2 acquire count=4
pool-1-thread-1 acquire count=7
pool-1-thread-1 release 7
pool-1-thread-2 release 4
pool-1-thread-3 acquire count=2
pool-1-thread-3 release 2

线程1和线程2会并发执行，因为两者的信号量和没有超过总信号量，当前两个线程释放掉信号量之后线程3才能继续执行.

源码分析:

1、构造函数 。

在构造函数中会初始化信号量值，这值最终是作为锁标志位state的值 。

?
1
Semaphore sem = new Semaphore( 12 ); //简单来说就是给锁标识位state赋值为12

2、Semaphore.acquire(n);简单理解为获取锁资源，如果获取不到线程阻塞 。

?
1
Semaphore.acquire(n); //从锁标识位state中获取n个信号量，简单来说是state = state-n 此时state大于0表示可以获取信号量，如果小于0则将线程阻塞
?
1
2
3
4
5
public void acquire( int permits) throws InterruptedException {
     if (permits < 0 ) throw new IllegalArgumentException();
     //获取锁
     sync.acquireSharedInterruptibly(permits);
   }

acquireSharedInterruptibly中的操作是获取锁资源，如果可以获取则将state= state-permits，否则将线程阻塞 。

?
1
2
3
4
5
6
7
public final void acquireSharedInterruptibly( int arg)
       throws InterruptedException {
     if (Thread.interrupted())
       throw new InterruptedException();
     if (tryAcquireShared(arg) < 0 ) //tryAcquireShared中尝试获取锁资源
       doAcquireSharedInterruptibly(arg); //将线程阻塞
   }

tryAcquireShared中的操作是尝试获取信号量值，简单来说就是state=state-acquires ，如果此时小于0则返回负值，否则返回大于新值，再判断是否将当线程线程阻塞 。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
protected int tryAcquireShared( int acquires) {
       for (;;) {
         if (hasQueuedPredecessors())
           return - 1 ;
       //获取state值
         int available = getState();
       //从state中获取信号量
         int remaining = available - acquires;
         if (remaining < 0 ||
           compareAndSetState(available, remaining))
         //如果信号量小于0则直接返回，表示无法获取信号量，否则将state值修改为新值
           return remaining;
       }
     }

doAcquireSharedInterruptibly中的操作简单来说是将当前线程添加到FIFO队列中并将当前线程阻塞.

?
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
/会将线程添加到FIFO队列中，并阻塞 
private void doAcquireSharedInterruptibly( int arg) 
     throws InterruptedException { 
     //将线程添加到FIFO队列中 
     final Node node = addWaiter(Node.SHARED); 
     boolean failed = true ; 
     try { 
       for (;;) { 
         final Node p = node.predecessor(); 
         if (p == head) { 
           int r = tryAcquireShared(arg); 
           if (r >= 0 ) { 
             setHeadAndPropagate(node, r); 
             p.next = null ; // help GC 
             failed = false ; 
             return ; 
           } 
         } 
         //parkAndCheckInterrupt完成线程的阻塞操作 
         if (shouldParkAfterFailedAcquire(p, node) && 
           parkAndCheckInterrupt()) 
           throw new InterruptedException(); 
       } 
     } finally { 
       if (failed) 
         cancelAcquire(node); 
     } 
   }

3、Semaphore.release(int permits)，这个函数的实现操作是将state = state+permits并唤起处于FIFO队列中的阻塞线程.

?
1
2
3
4
5
public void release( int permits) {
     if (permits < 0 ) throw new IllegalArgumentException();
   //state = state+permits，并将FIFO队列中的阻塞线程唤起
     sync.releaseShared(permits);
   }

releaseShared中的操作是将state = state+permits，并将FIFO队列中的阻塞线程唤起.

?
1
2
3
4
5
6
7
8
9
public final boolean releaseShared( int arg) {
     //tryReleaseShared将state设置为state = state+arg
     if (tryReleaseShared(arg)) {
       //唤起FIFO队列中的阻塞线程
       doReleaseShared();
       return true ;
     }
     return false ;
   }

tryReleaseShared将state设置为state = state+arg 。

?
1
2
3
4
5
6
7
8
9
10
11
protected final boolean tryReleaseShared( int releases) {
       for (;;) {
         int current = getState();
         int next = current + releases;
         if (next < current) // overflow
           throw new Error( "Maximum permit count exceeded" );
         //将state值设置为state=state+releases
         if (compareAndSetState(current, next))
           return true ;
       }
     }

doReleaseShared()唤起FIFO队列中的阻塞线程 。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private void doReleaseShared() { 
  
     for (;;) { 
       Node h = head; 
       if (h != null && h != tail) { 
         int ws = h.waitStatus; 
         if (ws == Node.SIGNAL) { 
           if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0 )) 
             continue ;      // loop to recheck cases 
           //完成阻塞线程的唤起操作 
           unparkSuccessor(h); 
         } 
         else if (ws == 0 && 
              !compareAndSetWaitStatus(h, 0 , Node.PROPAGATE)) 
           continue ;        // loop on failed CAS 
       } 
       if (h == head)          // loop if head changed 
         break ; 
     } 
   }

总结：Semaphore简单来说设置了一个信号量池state，当线程执行时会从state中获取值，如果可以获取则线程执行，并且在执行后将获取的资源返回到信号量池中，并唤起其他阻塞线程；如果信号量池中的资源无法满足某个线程的需求则将此线程阻塞.

Semaphore源码:

?
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
public class Semaphore implements java.io.Serializable {
   private static final long serialVersionUID = -3222578661600680210L;
   private final Sync sync;
   abstract static class Sync extends AbstractQueuedSynchronizer {
     private static final long serialVersionUID = 1192457210091910933L;
     //设置锁标识位state的初始值
     Sync( int permits) {
       setState(permits);
     }
     //获取锁标识位state的值，如果state值大于其需要的值则表示锁可以获取
     final int getPermits() {
       return getState();
     }
     //获取state值减去acquires后的值，如果大于等于0则表示锁可以获取
     final int nonfairTryAcquireShared( int acquires) {
       for (;;) {
         int available = getState();
         int remaining = available - acquires;
         if (remaining < 0 ||
           compareAndSetState(available, remaining))
           return remaining;
       }
     }
     //释放锁
     protected final boolean tryReleaseShared( int releases) {
       for (;;) {
         int current = getState();
         //将state值加上release值
         int next = current + releases;
         if (next < current) // overflow
           throw new Error( "Maximum permit count exceeded" );
         if (compareAndSetState(current, next))
           return true ;
       }
     }
     //将state的值减去reductions
     final void reducePermits( int reductions) {
       for (;;) {
         int current = getState();
         int next = current - reductions;
         if (next > current) // underflow
           throw new Error( "Permit count underflow" );
         if (compareAndSetState(current, next))
           return ;
       }
     }
     final int drainPermits() {
       for (;;) {
         int current = getState();
         if (current == 0 || compareAndSetState(current, 0 ))
           return current;
       }
     }
   }
   //非公平锁
   static final class NonfairSync extends Sync {
     private static final long serialVersionUID = -2694183684443567898L;
     NonfairSync( int permits) {
       super (permits);
     }
     protected int tryAcquireShared( int acquires) {
       return nonfairTryAcquireShared(acquires);
     }
   }
   //公平锁
   static final class FairSync extends Sync {
     private static final long serialVersionUID = 2014338818796000944L;
     FairSync( int permits) {
       super (permits);
     }
     protected int tryAcquireShared( int acquires) {
       for (;;) {
         if (hasQueuedPredecessors())
           return - 1 ;
         int available = getState();
         int remaining = available - acquires;
         if (remaining < 0 ||
           compareAndSetState(available, remaining))
           return remaining;
       }
     }
   }
   //设置信号量
   public Semaphore( int permits) {
     sync = new NonfairSync(permits);
   }
   public Semaphore( int permits, boolean fair) {
     sync = fair ? new FairSync(permits) : new NonfairSync(permits);
   }
   //获取锁
   public void acquire() throws InterruptedException {
     sync.acquireSharedInterruptibly( 1 );
   }
   public void acquireUninterruptibly() {
     sync.acquireShared( 1 );
   }
   public boolean tryAcquire() {
     return sync.nonfairTryAcquireShared( 1 ) >= 0 ;
   }
   public boolean tryAcquire( long timeout, TimeUnit unit)
     throws InterruptedException {
     return sync.tryAcquireSharedNanos( 1 , unit.toNanos(timeout));
   }
   public void release() {
     sync.releaseShared( 1 );
   }
   //获取permits值锁
   public void acquire( int permits) throws InterruptedException {
     if (permits < 0 ) throw new IllegalArgumentException();
     sync.acquireSharedInterruptibly(permits);
   }
   public void acquireUninterruptibly( int permits) {
     if (permits < 0 ) throw new IllegalArgumentException();
     sync.acquireShared(permits);
   }
   public boolean tryAcquire( int permits) {
     if (permits < 0 ) throw new IllegalArgumentException();
     return sync.nonfairTryAcquireShared(permits) >= 0 ;
   }
   public boolean tryAcquire( int permits, long timeout, TimeUnit unit)
     throws InterruptedException {
     if (permits < 0 ) throw new IllegalArgumentException();
     return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
   }
   //释放
   public void release( int permits) {
     if (permits < 0 ) throw new IllegalArgumentException();
     sync.releaseShared(permits);
   }
   public int availablePermits() {
     return sync.getPermits();
   }
   public int drainPermits() {
     return sync.drainPermits();
   }
   protected void reducePermits( int reduction) {
     if (reduction < 0 ) throw new IllegalArgumentException();
     sync.reducePermits(reduction);
   }
   public boolean isFair() {
     return sync instanceof FairSync;
   }
   public final boolean hasQueuedThreads() {
     return sync.hasQueuedThreads();
   }
   public final int getQueueLength() {
     return sync.getQueueLength();
   }
   protected Collection<Thread> getQueuedThreads() {
     return sync.getQueuedThreads();
   }
   public String toString() {
     return super .toString() + "[Permits = " + sync.getPermits() + "]" ;
   }
}

总结 。

以上就是本文关于Java并发编程Semaphore计数信号量详解的全部内容，希望对大家有所帮助。有什么问题，可以留言交流讨论。感谢朋友们对本站的支持！ 。

原文链接：http://blog.csdn.net/qq924862077/article/details/70224646 。

最后此篇关于Java并发编程Semaphore计数信号量详解的文章就讲到这里了,如果你想了解更多关于Java并发编程Semaphore计数信号量详解的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。

25 4 0
文章推荐： java 多线程的几种实现方法总结
文章推荐： ORACLE数据库中Rownum用法详解
文章推荐： Java微信跳一跳操作指南
文章推荐： Hadoop MultipleOutputs输出到多个文件中的实现方法
qq735679552
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
全站热门文章
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com