Java executors : 작업이 완료되면 차단하지 않고 알림을받는 방법?
실행기 서비스에 제출해야 할 작업으로 가득 찬 대기열이 있다고 가정 해보십시오. 한 번에 하나씩 처리하기를 원합니다. 내가 생각할 수있는 가장 간단한 방법은 다음과 같습니다.
- 대기열에서 작업 수행
- 유언 집행자에게 제출
- 반환 된 Future에서 .get을 호출하고 결과가 제공 될 때까지 차단
- 대기열에서 다른 작업을 수행하십시오 ...
그러나 완전히 차단하지 않으려 고합니다. 작업이 한 번에 하나씩 처리되어야하는 10,000 개의 대기열이있는 경우 대부분 차단 된 스레드를 유지하므로 스택 공간이 부족합니다.
내가 원하는 것은 작업을 제출하고 작업이 완료되면 호출되는 콜백을 제공하는 것입니다. 이 콜백 알림을 플래그로 사용하여 다음 작업을 보냅니다. (functionaljava와 jetlang은 분명히 그러한 비 차단 알고리즘을 사용하지만 코드를 이해할 수 없습니다)
내 실행 프로그램을 작성하는 데 부족한 JDK의 java.util.concurrent를 사용하여 어떻게 할 수 있습니까?
(이 작업을 제공하는 대기열은 자체적으로 차단 될 수 있지만 나중에 해결해야 할 문제입니다)
완료 알림에 전달할 매개 변수를 수신하도록 콜백 인터페이스를 정의하십시오. 그런 다음 작업이 끝날 때 호출하십시오.
Runnable 태스크를위한 일반 랩퍼를 작성하여에 제출할 수도 ExecutorService있습니다. 또는 Java 8에 내장 된 메커니즘에 대해서는 아래를 참조하십시오.
class CallbackTask implements Runnable {
private final Runnable task;
private final Callback callback;
CallbackTask(Runnable task, Callback callback) {
this.task = task;
this.callback = callback;
}
public void run() {
task.run();
callback.complete();
}
}
와 함께 CompletableFutureJava 8에는 프로세스를 비동기식 및 조건부로 완료 할 수있는 파이프 라인을 작성하는보다 정교한 수단이 포함되었습니다. 여기에 숙고되었지만 완전한 알림의 예가 있습니다.
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
public class GetTaskNotificationWithoutBlocking {
public static void main(String... argv) throws Exception {
ExampleService svc = new ExampleService();
GetTaskNotificationWithoutBlocking listener = new GetTaskNotificationWithoutBlocking();
CompletableFuture<String> f = CompletableFuture.supplyAsync(svc::work);
f.thenAccept(listener::notify);
System.out.println("Exiting main()");
}
void notify(String msg) {
System.out.println("Received message: " + msg);
}
}
class ExampleService {
String work() {
sleep(7000, TimeUnit.MILLISECONDS); /* Pretend to be busy... */
char[] str = new char[5];
ThreadLocalRandom current = ThreadLocalRandom.current();
for (int idx = 0; idx < str.length; ++idx)
str[idx] = (char) ('A' + current.nextInt(26));
String msg = new String(str);
System.out.println("Generated message: " + msg);
return msg;
}
public static void sleep(long average, TimeUnit unit) {
String name = Thread.currentThread().getName();
long timeout = Math.min(exponential(average), Math.multiplyExact(10, average));
System.out.printf("%s sleeping %d %s...%n", name, timeout, unit);
try {
unit.sleep(timeout);
System.out.println(name + " awoke.");
} catch (InterruptedException abort) {
Thread.currentThread().interrupt();
System.out.println(name + " interrupted.");
}
}
public static long exponential(long avg) {
return (long) (avg * -Math.log(1 - ThreadLocalRandom.current().nextDouble()));
}
}
Java 8에서는 CompletableFuture 를 사용할 수 있습니다 . 다음은 코드를 사용하여 사용자 서비스에서 사용자를 가져 와서 내보기 객체에 매핑 한 다음 내보기를 업데이트하거나 오류 대화 상자를 표시하는 예제입니다 (GUI 응용 프로그램).
CompletableFuture.supplyAsync(
userService::listUsers
).thenApply(
this::mapUsersToUserViews
).thenAccept(
this::updateView
).exceptionally(
throwable -> { showErrorDialogFor(throwable); return null; }
);
비동기 적으로 실행됩니다. 두 가지 개인 방법을 사용 mapUsersToUserViews하고 updateView있습니다 : 및 .
Guava의 청취 가능한 미래 API를 사용 하고 콜백을 추가하십시오. Cf. 웹 사이트에서 :
ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
ListenableFuture<Explosion> explosion = service.submit(new Callable<Explosion>() {
public Explosion call() {
return pushBigRedButton();
}
});
Futures.addCallback(explosion, new FutureCallback<Explosion>() {
// we want this handler to run immediately after we push the big red button!
public void onSuccess(Explosion explosion) {
walkAwayFrom(explosion);
}
public void onFailure(Throwable thrown) {
battleArchNemesis(); // escaped the explosion!
}
});
당신은 확장 할 수 FutureTask클래스를하고, 무시 done()하고 추가 방법을 FutureTask받는 개체를 ExecutorService소위, done()때 방법은 호출합니다 FutureTask즉시 완료.
ThreadPoolExecutor또한이 beforeExecute와 afterExecute당신이 무시하고 사용을 할 수 후크 방법. 여기에서 설명이다 ThreadPoolExecutor의 Javadoc과는 .
후크 방법
이 클래스는 각 작업 실행 전후에 호출 되는 보호 된 재정의
beforeExecute(java.lang.Thread, java.lang.Runnable)및afterExecute(java.lang.Runnable, java.lang.Throwable)메서드를 제공합니다 . 이들은 실행 환경을 조작하는 데 사용될 수 있습니다. 예를 들어, 다시 초기화ThreadLocals, 통계 수집 또는 로그 항목 추가. 또한terminated()일단Executor종료 된 후에 수행해야하는 특수 처리를 수행하기 위해 메소드 를 대체 할 수 있습니다 . 후크 또는 콜백 메소드에서 예외가 발생하면 내부 작업자 스레드가 실패하고 갑자기 종료 될 수 있습니다.
를 사용하십시오 CountDownLatch.
그것은 시작되었으며 java.util.concurrent계속하기 전에 여러 스레드가 실행을 완료 할 때까지 기다리는 방법입니다.
당신이 찾고있는 콜백 효과를 얻으려면 약간의 추가 작업이 필요합니다. 즉, 이것을 사용하고 CountDownLatch대기 하는 별도의 스레드에서 직접 처리 한 다음 통지해야 할 사항을 알리는 작업을 계속합니다. 콜백이나 그 효과와 유사한 것은 기본적으로 지원되지 않습니다.
편집 : 이제 귀하의 질문을 더 이해했기 때문에 불필요하게 너무 멀리 도달하고 있다고 생각합니다. 정기적 SingleThreadExecutor인 경우 모든 작업을 수행하면 기본적으로 큐잉이 수행됩니다.
작업이 동시에 실행되지 않도록하려면 SingleThreadedExecutor 를 사용하십시오 . 작업은 제출 된 순서대로 처리됩니다. 작업을 보류 할 필요조차 없으며 exec에 제출하십시오.
다음을 Callback사용하여 메커니즘 을 구현 하는 간단한 코드ExecutorService
import java.util.concurrent.*;
import java.util.*;
public class CallBackDemo{
public CallBackDemo(){
System.out.println("creating service");
ExecutorService service = Executors.newFixedThreadPool(5);
try{
for ( int i=0; i<5; i++){
Callback callback = new Callback(i+1);
MyCallable myCallable = new MyCallable((long)i+1,callback);
Future<Long> future = service.submit(myCallable);
//System.out.println("future status:"+future.get()+":"+future.isDone());
}
}catch(Exception err){
err.printStackTrace();
}
service.shutdown();
}
public static void main(String args[]){
CallBackDemo demo = new CallBackDemo();
}
}
class MyCallable implements Callable<Long>{
Long id = 0L;
Callback callback;
public MyCallable(Long val,Callback obj){
this.id = val;
this.callback = obj;
}
public Long call(){
//Add your business logic
System.out.println("Callable:"+id+":"+Thread.currentThread().getName());
callback.callbackMethod();
return id;
}
}
class Callback {
private int i;
public Callback(int i){
this.i = i;
}
public void callbackMethod(){
System.out.println("Call back:"+i);
// Add your business logic
}
}
산출:
creating service
Callable:1:pool-1-thread-1
Call back:1
Callable:3:pool-1-thread-3
Callable:2:pool-1-thread-2
Call back:2
Callable:5:pool-1-thread-5
Call back:5
Call back:3
Callable:4:pool-1-thread-4
Call back:4
주요 사항 :
- 당신이 FIFO 순서로 순서대로 처리 작업을하려는 경우, 교체
newFixedThreadPool(5)로newFixedThreadPool(1) callback이전 작업 의 결과를 분석 한 후 다음 작업을 처리하려면 아래 줄의 주석 처리를 제거하십시오.//System.out.println("future status:"+future.get()+":"+future.isDone());다음
newFixedThreadPool()중 하나로 대체 할 수 있습니다Executors.newCachedThreadPool() Executors.newWorkStealingPool() ThreadPoolExecutor사용 사례에 따라
콜백 메소드를 비동기 적으로 처리하려면
ㅏ. 공유 가능한
ExecutorService or ThreadPoolExecutor작업에 공유 전달비.
Callable분석법을Callable/Runnable작업으로 변환씨. 콜백 작업을
ExecutorService or ThreadPoolExecutor
Matt의 답변에 추가하기 위해 콜백을 사용하는 방법을 보여주는 더 많은 예제가 있습니다.
private static Primes primes = new Primes();
public static void main(String[] args) throws InterruptedException {
getPrimeAsync((p) ->
System.out.println("onPrimeListener; p=" + p));
System.out.println("Adios mi amigito");
}
public interface OnPrimeListener {
void onPrime(int prime);
}
public static void getPrimeAsync(OnPrimeListener listener) {
CompletableFuture.supplyAsync(primes::getNextPrime)
.thenApply((prime) -> {
System.out.println("getPrimeAsync(); prime=" + prime);
if (listener != null) {
listener.onPrime(prime);
}
return prime;
});
}
출력은 다음과 같습니다.
getPrimeAsync(); prime=241
onPrimeListener; p=241
Adios mi amigito
구아바를 사용하여 Pache의 답변을 확장 한 것 ListenableFuture입니다.
In particular, Futures.transform() returns ListenableFuture so can be used to chain async calls. Futures.addCallback() returns void, so cannot be used for chaining, but is good for handling success/failure on an async completion.
// ListenableFuture1: Open Database
ListenableFuture<Database> database = service.submit(() -> openDatabase());
// ListenableFuture2: Query Database for Cursor rows
ListenableFuture<Cursor> cursor =
Futures.transform(database, database -> database.query(table, ...));
// ListenableFuture3: Convert Cursor rows to List<Foo>
ListenableFuture<List<Foo>> fooList =
Futures.transform(cursor, cursor -> cursorToFooList(cursor));
// Final Callback: Handle the success/errors when final future completes
Futures.addCallback(fooList, new FutureCallback<List<Foo>>() {
public void onSuccess(List<Foo> foos) {
doSomethingWith(foos);
}
public void onFailure(Throwable thrown) {
log.error(thrown);
}
});
NOTE: In addition to chaining async tasks, Futures.transform() also allows you to schedule each task on a separate executor (Not shown in this example).
You may use a implementation of Callable such that
public class MyAsyncCallable<V> implements Callable<V> {
CallbackInterface ci;
public MyAsyncCallable(CallbackInterface ci) {
this.ci = ci;
}
public V call() throws Exception {
System.out.println("Call of MyCallable invoked");
System.out.println("Result = " + this.ci.doSomething(10, 20));
return (V) "Good job";
}
}
where CallbackInterface is something very basic like
public interface CallbackInterface {
public int doSomething(int a, int b);
}
and now the main class will look like this
ExecutorService ex = Executors.newFixedThreadPool(2);
MyAsyncCallable<String> mac = new MyAsyncCallable<String>((a, b) -> a + b);
ex.submit(mac);
'Programing' 카테고리의 다른 글
| C # : 상속 된 이벤트 발생 (0) | 2020.06.22 |
|---|---|
| LINQ 식의 String.IsNullOrWhiteSpace (0) | 2020.06.22 |
| jQuery : 텍스트 상자 내에서 Tab 키 누르기를 캡처하는 방법 (0) | 2020.06.22 |
| py2exe-단일 실행 파일 생성 (0) | 2020.06.22 |
| Git없이 파일에 Git SHA1을 할당하는 방법은 무엇입니까? (0) | 2020.06.22 |