자바 병렬 프로 그래 밍 Semaphore 의 응용 및 원본 분석

27908 단어 자바
What
Semaphore 표지 신 호 량, 지 정 된 수량의 라인 이 특정한 자원 에 동시에 접근 할 수 있 도록 합 니 다
How
다음 두 부 를 통 해 신 호 량 을 실현 합 니 다.
  • acquire 방법 은 접근 허가 (허 가 를 받 지 못 하면 스 레 드 석방 허가 가 있어 허 가 를 받 을 때 까지 기다린다)
  • release 접근 허가 석방
  • 응용 장면
  • 특정한 자원 탱크 제한 을 실현 하고 데이터 베이스 연결 탱크
  • 와 유사 하 다.
  • 용기 에 경 계 를 가 한다. 예 를 들 어 하나의 집합 에 최대 5 개의 요소 만 추가 할 수 있다
  • .
  • 자원 동시 방문 수량 제한
  • 일반적인 자물쇠 로 사용 (신 호 량 이 1 일 경우 일반적인 자물쇠 신 호 량 이 1 일 때 공유 자물쇠 보다 많 음)
  • Semaphore 코드 예시
    import java.util.concurrent.Semaphore;
    
    public class SemaphoreDemo implements Runnable {
         
    
        Semaphore semaphore = new Semaphore(5);
    
        public static void main(String[] args) {
         
            SemaphoreDemo semaphoreDemo = new SemaphoreDemo();
            for (int i = 0; i < 10; i++) {
         
                Thread thread = new Thread(semaphoreDemo);
                thread.start();
            }
        }
    
        @Override
        public void run() {
         
            try {
         
                //       (        ,     ,                 )
                semaphore.acquire();
                Thread.sleep(1000);
                System.out.println(System.currentTimeMillis() + ", " + Thread.currentThread().getName() + ",     !");
                //       
                semaphore.release();
            } catch (InterruptedException e) {
         
                e.printStackTrace();
            }
        }
    }
    

    예 를 들 어 10 개의 스 레 드 가 신 호 량 이 5 인 자원 을 경쟁 적 으로 사용 하지만 매번 5 개의 스 레 드 만 허 가 를 받 고 다른 스 레 드 는 허 가 를 받 은 스 레 드 방출 허 가 를 기 다 려 야 허 가 를 받 을 수 있다.
    Semaphore 소스 코드 분석
    관건 적 인 방법 은 다음 과 같다.
  • 구조 방법: new Semaphore (5);
  • 획득 허가: semaphore. acquire ();
  • 석방 허가: semaphore. release ();

  • 이어서 우 리 는 이 세 가지 방법 에 착안 하여 소스 코드 분석 을 진행 할 것 이다.
    1. 구조 방법: new Semaphore (5)
        public Semaphore(int permits) {
         
            sync = new NonfairSync(permits);
        }
     
        public Semaphore(int permits, boolean fair) {
         
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }
    

    기본 구조 방법 은 불공 정 공유 자물쇠 로 구조 적 매개 변수 fair 를 통 해 공평 하거나 불공 정 한 것 을 선택 할 수 있 습 니 다. ReentantLock 과 유사 합 니 다.
    2. 획득 허가: semaphore. acquire ()
        public void acquire() throws InterruptedException {
         
            //      AQS     
            sync.acquireSharedInterruptibly(1);
        }
    

    AQS acquireSharedInterruptibly 방법 을 호출 합 니 다.
        public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
         
            if (Thread.interrupted()) {
         
                throw new InterruptedException();
            }
            if (tryAcquireShared(arg) < 0) {
         
                doAcquireSharedInterruptibly(arg);
            }
        }
    

    그 중에서 tryAcquireShared Sync 실현 에 의존 하고 Semaphore 에 AQS 의 Sync 실현 클래스 가 있 습 니 다. 방법 은 다음 과 같 습 니 다.
            //        
            protected int tryAcquireShared(int acquires) {
         
                for (; ; ) {
         
                    //             -1
                    if (hasQueuedPredecessors())
                        return -1;
                    int available = getState(); //       
                    int remaining = available - acquires; //       
    
                    if (remaining < 0 || compareAndSetState(available, remaining))
                        //        
                        return remaining;
                }
            }
    

    이것 은 FairSync 의 try AcquireShared 방법 으로 NonfairSync 에서 hasQueuedPredecessors() 판단 이 없고 나머지 는 같다.
    방법 에서 알 수 있 듯 이 최종 적 으로 돌아 온 것 은 남 은 허가 수량 으로 다음 과 같은 몇 가지 상황 이 있다.
  • 남 은 허가 수량 < 0 이 있 으 면 doAcquireSharedInterruptibly 방법 을 실행 하여 스 레 드 를 자전 적 으로 기다 리 게 합 니 다. 여 기 는 다른 스 레 드 가 허 가 를 내 린 후에 스 레 드 가 깨 어 나 서 얻 으 려 고 시도 합 니 다
        private void doAcquireSharedInterruptibly(int arg)
                throws InterruptedException {
           
            //         
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
           
                for (; ; ) {
            //   
                    final Node p = node.predecessor();
                    if (p == head) {
            //           AQS         AQS          
                        int r = tryAcquireShared(arg); //        
                        if (r >= 0) {
           
                            setHeadAndPropagate(node, r); //     
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    //            
                    if (shouldParkAfterFailedAcquire(p, node) &&
                            parkAndCheckInterrupt()) {
           
                        throw new InterruptedException();
                    }
                }
            } finally {
           
                if (failed) {
           
                    cancelAcquire(node); //           
                }
            }
        }
    
  • 그렇지 않 으 면 허가 수량 을 받 고 계속 정상적으로 집행 하 며 막 히 지 않 는 다
  • 3. 석방 허가: semaphore. release ()
        public void release() {
         
            sync.releaseShared(1);
        }
    

    마찬가지 로 AQS releaseShared 방법 을 호출 했 습 니 다. 코드 를 보 세 요.
        public final boolean releaseShared(int arg) {
         
            //   AQS    tryReleaseShared
            if (tryReleaseShared(arg)) {
         
                //          
                doReleaseShared();
                return true;
            }
            return false;
        }
    
    tryReleaseShared 하위 클래스 동기 화 에 의 해 이 루어 집 니 다. 코드 는 다음 과 같 습 니 다.
            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");
                    if (compareAndSetState(current, next)) // CAS          
                        return true;
                }
            }
    

    석방 허가 성공 시 AQS doReleaseShared 방법 을 계속 호출 하여 후속 노드 를 깨 우 면 허 가 를 얻 을 수 있 습 니 다.
        private void doReleaseShared() {
         
            for (; ; ) {
          //     
                Node h = head;
                //                  
                if (h != null && h != tail) {
         
                    int ws = h.waitStatus;
                    if (ws == Node.SIGNAL) {
         
                        //   status 0
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) {
         
                            continue; //     
                        }
                        //          
                        unparkSuccessor(h);
                    } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
          //
                        continue; //        
                    }
                }
                if (h == head) {
         
                    break;
                }
            }
        }
    

    총결산
    Semaphore 는 신 호 량 을 저장 하기 위해 AQS 동기 화 상 태 를 사용 합 니 다.acquireSharedInterruptibly 계수 (허가 획득) 를 줄 일 수 있 습 니 다. 계수 가 비정 상 일 때 스 레 드 를 막 습 니 다. 그렇지 않 으 면 스 레 드 releaseShared 방법 이 계수 (허가 방출) 를 증가 시 키 고 신 호 량 제한 을 초과 하지 않 을 때 스 레 드 의 차단 을 해제 합 니 다 (허 가 를 받 은 스 레 드).

    좋은 웹페이지 즐겨찾기