스크립트 개발 / 스레드 풀 `DFF.THREAD`

DFF.THREAD는 단일 Task 내에서 IO 집약적 함수(예: 대량 HTTP 요청)를 동시에 실행하는 데 사용됩니다. 스레드 풀은 DataFlux Func가 관리합니다.

API

메서드 / 속성	설명
`pool_size`	명시적으로 구성된 스레드 풀 크기; 아직 설정되지 않은 경우 `None`
`set_pool_size(pool_size)`	스레드 풀 크기를 설정하며, 첫 `submit(...)` 호출 전에 호출해야 합니다
`submit(fn, args, *kwargs)`	함수를 제출하고 결과 Key를 반환합니다
`get_result(key, wait=True)`	지정된 결과를 가져옵니다
`get_all_results(wait=True)`	전체 결과를 가져옵니다
`pop_result(wait=True)`	완료된 결과 하나를 꺼냅니다. 꺼낸 후에는 다른 메서드에서 다시 반환되지 않습니다
`is_all_finished`	전부 실행 완료되었는지 여부
`wait_all_finished()`	전부 실행 완료될 때까지 대기합니다

결과 객체 유형은 DFFThreadResult이며, 다음 속성을 포함합니다:

속성	설명
`key`	`submit(...)`가 반환한 결과 Key
`value`	함수 반환값; 실행 실패 시 일반적으로 `None`
`error`	함수에서 발생한 예외; 실행 성공 시 `None`

set_pool_size(...)는 구성 파일 config.yaml에 설정된 최댓값을 기준으로 검증되며, 첫 제출 이후 다시 설정해도 적용되지 않습니다. 명시적으로 설정하지 않으면 스레드 풀 생성 시 런타임 기본값이 사용됩니다.

예시

대량 HTTP 요청
import requests

def fetch(url):
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return resp.text

@DFF.API('대량 요청')
def batch_fetch(urls):
    DFF.THREAD.set_pool_size(10)

    for url in urls:
        DFF.THREAD.submit(fetch, url)

    return [
        {'key': result.key, 'error': repr(result.error)}
        if result.error
        else {'key': result.key, 'value': result.value}
        for result in DFF.THREAD.get_all_results()
    ]

결과 읽기 동작

get_result(key, wait=False) 및 pop_result(wait=False)는 매칭되는 준비된 결과가 없으면 None을 반환합니다.
함수를 하나 이상 제출한 후에만 get_all_results(...)를 호출할 수 있습니다.
get_all_results(...)는 완료 순서가 아니라 제출 Key 순서대로 결과를 반환합니다.
완료되는 대로 처리해야 할 때는 pop_result(wait=not DFF.THREAD.is_all_finished)를 호출할 수 있습니다.
pop_result(...)는 서로 독립적인 작업에 적합하며, get_all_results(...)는 모두 완료된 후 일괄 처리할 때 적합합니다.

Warning

모든 result.error를 확인해야 합니다. 그렇지 않으면 스레드 예외가 무시됩니다. 결과를 아직 수집하지 않았더라도 Task는 제출된 스레드 작업이 중지된 후에야 종료되므로, 각 스레드 작업에는 명확한 타임아웃 또는 다른 실행 경계를 설정해야 합니다.

스레드 풀은 IO 집약적 작업에 적합합니다. CPU 집약적 작업은 일반적으로 뚜렷한 속도 향상을 얻지 못합니다.

스크립트 개발 / 스레드 풀 DFF.THREAD

API

예시

결과 읽기 동작

스크립트 개발 / 스레드 풀 `DFF.THREAD`