1/9 자바 병렬 프로그래밍 4

- CompletionService는 Executor의 기능과 BlockingQueue의 기능을 하나로 모은 인터페이스이다. 완료된 결과값을 쌓아둘 블로킹큐를 생성하고. 작업이 완료된 순서대로 이 큐에 쌓이게 된다. 

 <중단정책>

- interrupt : 해당 스레드에게 중지를 요청하는것 . 이 요청을 받은 스레드는 스스로 적절하게 종료할수있게끔 행동해야한다

- 가장 기본적인 취소정책은 cancel메소드 호출 -> volatile로 선언한 취소플래그를 true로 변경-> 작업 종료 하는 것이다

- 그러나 블로킹큐의 put , take 같은 블로킹메서드를 호출하는 경우 반복문 작업 내부에서 취소요청이 들어왔는지 확인하지 못하는 경우가 생길 수 있다. 이런 경우 while 반복문의 조건확인 부분에서 인터럽트 여부를 직접확인 하는 방법으로 응답속도를 개선할 수 있다. 

<인터럽트에 대한 대응>

- 호출 스택의 상위메소드로 예외를 던져서 상위메소드에서 처리할수있도록 넘긴다. (throws InterruptException 한다)

- 호출 스택의 상단에 위치한 메소드가 직접 처리할 수 있도록 인터럽트 상태를 유지한다. ( Thread.currentThread.interrupt()를 하여 인터럽트 상태 유지. 상위 메서드가 얘를 관리할수있도록 한다.)

- 스레드 풀에 들어있는 스레드에 함부로 인터럽트를 걸면 안된다. 해당 스레드에 인터럽트가 걸리는 시점에 어떤 작업을 실행하고 있을지 알수없기 때문이다. 따라서 작업을 중단하려 할 때는 항상 스레드에 직접 인터럽트를 거는 대신 Future의 cancel메소드를 사용해야 한다. 

<스레드 기반 서비스 중단>

- 스레드를 소유하는 객체는 대부분 해당 스레드를 생성한 객체라고 볼 수 있다. 스레드 풀을 예로 들면, 스레드 풀에 들어있는 모든 작업 스레드는 해당하는 스레드 풀이 소유한다고 볼 수 있고, 따라서 개별 스레드에 인터럽트를 걸어야 하는 상황이 된다면 그 작업은 스레드를 소유한 스레드 풀에서 책임을 줘야 한다. 

- 애플리케이션이 직접 개별스레드에 액세스하는 대신 스레드 기반 서비스가 스레드의 시작부터 종료까지 모든 기능에 해당하는 메서드를 직접 제공해야 한다. 그러면 애플리케이션이 스레드 기반 서비스만 종료시키면 스레드 기반 서비스는 스스로가 소유한 모든 작업 스레드를 종료시키게 된다. 좋은 예로 ExecutorService인터페이스는 shutdown메소드와 shutdownNow메소드를 제공한다. 

<비정상적인 스레드 종료 상황 처리>

-예상할 수없었던 예외로 인하여 스레드가 비정상적으로 종료되는 경우 . 자바에서 기본적으로 제공하는 스레드풀에서는 작업에서 예상치 못한 예외가 발생했을때 해당 스레드가 종료되도록 하면서, try-finally구문을 사용해 스레드가 종료되기 전에 스레드 풀에 종료된다는 사실을 알려 다른 스레드를 대체해 실행할 수 있도록 하고 있다. 

- UncaughtExceptionHandler를 디폴트로 셋팅해주거나. 혹은 afterExecute메소드를 활용한다.

- 단 Callable로 작업결과를 받는 경우 ExecutionException에 감싸여진 상태로 예외가 넘어오므로 여기에서 처리해주어야 한다. 

<JVM종료>

- ExecutorService를 컴포지션으로 사용하는 클래스의 start()부분에 종료훅을 등록하여 자원을 모두 해제한후 stop할수있도록 한다. 

- 종료 훅이 여러개 등록되어 있는 경우에는 여러개의 종료 훅이 서로 동시에 실행 되기 때문에 다른 종료 훅에서 해당 서비스를 사용하고 있었다면(즉 힙메모리에서 해당 객체를 공유하고 있었다면) 이미 다른 곳에서 종료되어 문제가 발생할 수 있다. 이런 경우를 예방 하려면. 서비스별로 종료훅을 등록하기 보다는 모든 서비스를 정리할 수 있는 하나의 종료 훅을 사용해 각 서비스를 의존성에 맞춰 순서대로 정리하는 것도 방법이다. 실제로 서비스 간의 종속성이 명확히 보이는 어플리케이션이라면 종료시점의 마무리 절차를 순차적으로 처리하도록 하여 올바른 순서대로 서비스를 종료하고 마무리할 수 있다. 

 - 데몬 스레드는 예고 없이 종료될 수 있기 때문에 애플리케이션 내부에서 시작시키고 종료시키며 사용하기에는 그다지 좋은 방법이 아니다. 

---

<스레드부족 데드락> Thread Starvation Deadlock

- 특정 자원을 확보하고자 계속해서 대기하거나 풀 내부의 다른 작업이 실행되어야 알 수 있는 조건이 만족하기를 기다리는 것처럼 끝없이 대기할 가능성이 있는 기능을 사용하는 작업이 풀에 등록된 경우 발생할 수 있다. 

- 다른 작업에 의존성을 가지고 있는 작업을 실행시키는 경우.

- 스레드 풀의 크기는 직접적으로 지정하는 것 이외에도 스레드 풀에서 필요로 하는 자원이 제한되어 원하는 크기보다 작은 수준에서 동작하는 경우도 있다. 예를 들어 10개짜리 JDBC커넥션풀을 사용해야 한다면. 결국 실제로 수행될 수 있는 양은 JDBC풀의 크기인 10개에 불과한 셈이다.

---

- 적절한 스레드 풀 사이즈

- 집중대응정책 : 적절한 큐 사이즈 조정 및 거부정책 

- 스레드 팩토리 : 직접작성하여 커스터마이징할 수 있다. 스레드 이름을 의미있게 짓는다거나 로그기능을 포함하는 스레드를 넘긴다거나..

-----

<데드락>

- 상대가 자원을 놓기만을 기다리는 상태

- JVM은 데드락을 스스로 해결해주지못한다

- 데이터베이스 시스템의 경우 데드락상황에서 복구하는 기능을 가지고있다. (한쪽을 희생양으로 해서 트랜잭션을 강제종료)

- 락의 순서를 제어하여 데드락을 방지한다.

- 암묵적인 락 대신에 명시적인 락을 사용하고, 타임아웃을 지정하여 락을 확보하려는 시점에서 시간제한이 걸리면 이미 확보했던 락을 풀어주고 잠시 기다리다가 다시 작업을 시도해볼수있다.

<오픈호출>

- 락을 전혀 확보하지 않은 상태에서 메소드를 호출하는 것

- 데드락을 미연에 방지하고자 사용할 수 있다

- open call

<라이브 락>

- 특정작업의 결과를 받아와야 다음 단계로 넘어갈 수 있는작업이 실패할수밖에 없는 기능을 계속해서 재시도하는 경우에 볼수있다.

- 메시지를 제대로 전송하지 못했을 때 해당 전송 트랜잭션을 롤백하고 실패한 메시지를 큐의 맨 뒤에 쌓아두는 트랜잭션 메시지 전송 어플리케이션에서 자주 나타난다. 특정 메시지를 큐에서 뽑아 핸들러에 넘겼을때 핸들러는 같은 에러 결과를 내면서 계속해서 호출된다.

- 에러를 너무 완벽하게 처리하고자 회복 불가능한 오류를 회복 가능하다고 판단해 계속해서 재시도 하는 과정에서 나타난다. 

1/9 자바 병렬 프로그래밍 3

 - 프로듀서 컨슈머 패턴: 블로킹 큐를 이용하면 좀더 편하게 구현할 수 있다

재사용성이 높아지고 성능의 측면에서도 이득을 볼 수 있다. 프로듀서가 디스크, 네트워크 i/o에 시간을 많이 소비하고 컨슈머가 cpu를 많이 소비한다면- 둘 클래스로 분리, 단일 스레드에서 순차적으로 실행하는 것보다 성능이 크게 높아질 수 있다. 

- 객체풀: 풀 내부에 소유하고 있던 객체를 외부에 공개할 때 적절한 동기화 작업이 되어있고, 그와 함께 풀에서 객체를 빌려다 사용하는 스레드 역시 외부에 공개하지 않는다. 적절하게 소유권이 이전되며. 새로운 스레드 내부에 객체가 완전히 한정된다. 

- 작업 가로채기 패턴 : 컨슈머가 각자 덱을 가지며, 자신의 덱이 비었을 경우 다른 컨슈머의 덱을 살펴보고 맨 뒤에있는 작업을 자기 덱으로 가져온다. 맨뒤에있는 것을 가져오기때문에 해당 덱의 원 소유자와 경쟁이 생기지 않으며 규모가 큰 시스템에서 적절하다 . 컨슈머가 여러가지 작업을 해야 하는 경우 작업에 작업이 꼬리를 물며 쌓일수있는데 이러한 경우에 적합하다. 

- 블로킹: i/o를 기다리거나, 락이 해제되기를 기다리거나, sleep등에 걸리거나 -> waiting, blocked , timed-waiting 상태에 들어간다. 

- interrupt : 여러 스레드가 협력하며 작업하기 위한 방법. 강제로 멈추는 것이 아니라 중단을 '요청'한다. 

- 세마포어 활용 : connection풀 관리 - 컬렉션의 크기에 제한을 두고 . 남은 객체가 없을 때 다른 스레드가 확보했던 객체를 반납받아 사용할 수 있을 때까지 대기하도록 할수있다. 

- BoundedHashset : 컬렉션 크기가 가질 수 있는 최대 크기로 세마포어를 초기화 한다. add하기 전에 acquire하여 추가할 여유가 있는지 확인하고 add 해본다 . 만약 추가에 실패했다면 realease한다. 비슷하게 , remove는 remove하기 전에 release하여 세마포어에 퍼밋을 반납하여 남은 공간에 객체를 추가할 수 있도록 해준다. 

1/9 자바 병렬 프로그래밍 2

 - 스레드 안전성 확보 방법

1) 스레드 한정 : 

 -ThreadLocal 이용하기 . 

 - 스택한정(지역변수로만 사용)

2) 변경불가능한 객체로 만든다. 불변객체

- 모든 변수를 final로 선언

- 적절하게 초기화한다 ( 생성자에서 this를 노출하지 않는다. ) 

- 변경할수있는 메서드 제공x

* 값을 읽을때 -> 새로운 불변객체를 만들어 리턴한다

* 값을 변경할때 -> 새로운 불변객체를 만들어 리턴한다

읽을때와 변경할때 둘다 단일연산이 되어야한다

volatile은 변경하는 스레드가 한개일 경우에만 유효하게 된다

3) 동기화 한다.

- 올바르게 초기화해야한다. 쓰레기 혹은 잔여값이 들어있다. 자바의 경우 모든객체가 Object를 상속하므로 여기에 따른 쓰레기값이 들어있을수있다. 

- 어떤 객체의 상태 state 조합. 은 해당 객체가 가지고 있는 변수의 개수 * 그 변수의 상태범위.

- 불변객체의 경우 한번 넣은 변수값이 바뀌지 않으므로 상태범위가 1개이다. 따라서 올바른상태인지 올바르지 않은 상태인지 검증하기 매우쉬워진다

- 상태범위가 좁아질수록 올바르지 않은 상태인지 검증하기 쉽다

- 올바르지 않은 상태를 가질수있다면( 다른 스레드에 의해 값이 변경되거나. ) 반드시 단일연산처럼 동작하도록 동기화해야한다

- 만약 여러개의 변수를 통하여 객체의 상태를 정의한다면 그 여러개의 변수를 변경하는 과정 전체를 단일연산으로 동작하도록 만들어야 한다. 개개의 AtomicInteger는 단일연산을 지원하지만 그 각기 여러개를 변수로 가지고 있는 클래스라면 하나의 변수를 바꾸고 다른 변수를바꾸려는 과정에서 다른 스레드가 끼어들 수 있다는 것이다. 

- 객체의 상태는 해당 객체에 포함되는 모든 객체와 변수가 가질 수 있는 전체 상태의 부분집합이다. 

- 객체 자체의 암묵적인 락을 사용하여 동기화할 수도 있고, 락으로 활용하기 위한 private객체를 준비하여 락으로 사용할 수 있다. 

만약 이 객체를 외부에서 접근할수있는 수단을 제공한다면 함께 동기화 작업에 참여할수도있다. 

- 스레드 안전한 객체 : 복사본을 제공한다.

성능에 문제가 될수있다(매번 새로운 객체 생성) , 요구사항에 따라서는 적합하지 않을 수 있다. 시시각각으로 변하는 데이터를 받기를 원하는 경우. 

- 각각의 변수가 모두 스레드 안전한 클래스라고 할지라도 전체적으로는 스레드 안전성을 잃을 수 있다. 내부 변수간에 의존성을 가지고 있는 경우가 그러하다.

- 기능추가 : 1) 해당 라이브러리 직접수정 2) extends이용. 그러나 해당 라이브러리가 변경될경우 쥐도새도모르게 내 코드의 안전성이 파괴될 위험이 있다. 

3) 도우미 클래스작성 -> composition을 활용한다- 위험!

4) 도우미 클래스 작성, implements를 활용한다

1/9 자바 병렬 프로그래밍

 -재진입성 (reentrant): 동기화 메서드는 스레드 단위로 진입한다. 요청 단위가 아니다

- volatile 키워드 : 이 변수는 공유변수이며 재배치되서는 안된다는 뜻. 캐싱되지않는다

가시성만 보장한다

연산의 단일성은 보장하지 않음(약하다)

- 락은 연산의 단일성 및 가시성 둘다 보장

 Situations when constructors are not thread safe are for example :

• modifying static variable from within a constructor without proper synchronization mechanism
• publishing this reference from within constructor before the object is fully initialized and can be accessed by other threads.

https://www.baeldung.com/java-thread-constructor

- 완전히 생성된 후에 그 객체를 다룰수있도록 해야한다

1/6 카프카 컨슈머 그룹 테스트

 

파티션 3으로 나누었음

데이터가 파티션별로 슉슉 들어간다. 

컨슈머가 꺼져있는 동안 lag가 발생하는것확인할수있음

더불어 다시 키면- 파티션전부 subscribe하고있었기 때문에 순서 보장없이 막들어온다

순서를 보장하려면?? 파티션을 1개만 만들고.. 할당한다!

1/6 로컬에 우분투 설치, 카프카 실험

  

./bin/kafka-console-consumer.sh --bootstrap-server 13.124.113.23:9092 -topic test-02 --from-beginning

./bin/kafka-console-producer.sh --bootstrap-server 13.124.113.23:9092 --topic test-02

1/6 aws에 카프카 설치, 로컬에서 실험

 

.\bin\windows\kafka-console-consumer.bat --bootstrap-server 13.124.113.23:9092 --topic test-01 --from-beginning

.\bin\windows\kafka-console-producer.bat --bootstrap-server 13.124.113.23:9092 --topic test-01