재사용성이 뛰어난 배치처리 시스템을 PHP를 통해 간단하게 구현해봅시다
안녕하세요, 라이언 입니다!
개발 중에 배치처리를 복잡하게 진행하여야 하는 경우,
재사용성을 높이는 모듈방식으로 진행하는 방식을 소개하겠습니다!
핵심 명칭이나, 핵심 아이디어는 Java Spring에서 사용되는 Spring Batch를 활용하여 구현하였습니다.
그만큼 Spring Batch의 구조는 정말 확장성이 높고 실용성이 뛰어나다고 생각합니다.
현재 배치처리가 필요한 프로젝트는 PHP 7.4 버전을 사용하고 있습니다.
PHP는 객체지향이 생겨난지 오랜 역사가 없기 때문에, 객체지향을 최대한 활용해야 하는 상황에서는,
PHP 버전을 미리 파악하는 것이 매우 중요한 시작이라고 볼 수 있습니다.
PHP 7.4에서도 웬만한 객체지향은 가능하지만, 딱 하나 문제점이, Generic을 지원하지 않는다는 부분이 있었습니다.
이는 중간에 Array를 사용하게 될 경우, 내부 타입을 명확히 지정할 수 없다는 아쉬움이 존재합니다.
물론, 배열을 간접적으로 접근하여, 생성자나 내부 메서드에서만 다룬다면, 타입검증으로 이를 충분히 보완하는 것이 가능합니다.
전체적인 배치처리 구조 흐름 설계
저는 전체적으로 하나의 배치를 Job으로 정의하였습니다. 여러개의 배치처리를 돌린다면, Job을 여러번 정의하는 구조가 되도록 하였습니다.
Spring Batch에서는 Step을 통해서, 하나의 Job에 여러 개의 배치를 구성할 수 있는 구조로 되어있으나,
해당 방식은 여러 개의 Job을 돌리는 것으로 대안이 있기 때문에, 고려하지 않았습니다.
단순히 목적은, 하나의 배치처리를 모듈화하여, 조건부 처리를 가능하게 하는 것이 목적이었습니다.
Job이 정의되고, Job이 시작되면, JobListener를 먼저 호출하게 됩니다.
JobListener의 역할은 배치 작업이 시작되기 전/후 에 필요한 사전/마무리 작업을 진행하는 역할을 진행합니다.
JobListener에 최초에 진입하면, 배치처리를 시작하기 전에 필요한 사전 데이터 로드 작업을 주로 수행합니다. 캐싱(Caching) 과정을 주로 수행합니다.
캐싱 작업이 모두 완료되면, 이제 ItemReader를 통해서, 배치처리될 데이터를 조회합니다.
ItemReader는 단순히 데이터를 조회하는 역할만 수행합니다.
ItemProcessor는 ItemReader로 조회된 데이터 중 하나의 Item을 순차적으로 받아서,
데이터 변환 과정을 지정합니다. 여기서 말하는 데이터 변환 과정은 Item의 특징에 맞게 조건부 처리하여,
DB에 최종적으로 적재되기 편한 형태로 가공처리를 수행하는 것을 의미합니다.
가공 처리된 Item은 ItemWriter 과정에서 실제로 DB에 데이터를 추가하게 됩니다.
마무리로, 다시 JobListener로 돌아와서, 배치작업 처리 후, 필요한 마무리 작업을 수행합니다.
배치 결과에 따른 데이터 결과 저장과 같은 역할이 들어갈 수 있겠습니다.
전체적인 흐름은 정말로 매우 단순합니다.
원래 배치처리를 단순하게 흘러가야 합니다. 특히 중요한 데이터를 처리하는 로직일수록 그 중요성은 매우 커집니다.
코드로 한 번 살펴볼까요?
<?php
class Job {
private JobListener $jobListener;
private ItemReader $itemReader;
private ItemProcessor $itemProcessor;
private ItemWriter $itemWriter;
private array $jobContext = array();
private $jf;
private bool $withTransaction = false;
public function __construct(JobListener $jobListener, itemReader $itemReader, itemProcessor $itemProcessor, itemWriter $itemWriter, $param) {
$this->jobListener = $jobListener;
$this->itemReader = $itemReader;
$this->itemProcessor = $itemProcessor;
$this->itemWriter = $itemWriter;
$this->jf = JF::getInstance();
$this->withTransaction = isset($param['with_transaction']) && $param['with_transaction'] === 't';
}
/**
* Job을 결과를 확인하기 위한, Context 데이터를 가져옵니다.
* @return array
*/
public function getJobContext() {
return $this->jobContext;
}
/**
* 작업을 실행한다.
* @param $param, 작업을 수행할 떄, 필요한 JobParameter를 주입합니다.
* @return void
*/
public function run($param) {
$transactionId = null;
// 트랜잭션 시작.
if( $this->withTransaction ) {
$transactionId = $this->jf->db->begin();
}
// 외부에서 주입된 데이터를 항상 어느 프로세스에서도 사용가능하게, 저장
$this->jobContext['job_parameter'] = $param;
// 작업을 실행하기 전에, 사전 작업 처리.
if (isset($this->jobListener)) {
$this->jobListener->beforeJob($this->jobContext, $param);
}
if ($this->checkAndStopJob($transactionId)) {
return;
}
try {
$list = $this->itemReader->read($this->jobContext);
// item 하나당, processor => writer 로직을 태우도록 처리
foreach($list AS $key => $value) {
$item = $this->itemProcessor->process($value, $this->jobContext);
$this->itemWriter->write($item, $this->jobContext);
}
} catch (Exception $e) {
if ($this->withTransaction) {
$this->jf->db->rollback($transactionId);
exit;
}
}
// 작업 실행 후, 이후 처리.
if (isset($this->jobListener)) {
$this->jobListener->afterJob($this->jobContext, $param);
}
// 트랜잭션 커밋.
if( $this->withTransaction && $transactionId !== null ) {
$this->jf->db->commit($transactionId);
}
}
private function checkAndStopJob($transactionId = null) {
if ($this->jobContext['job_parameter']['stop_job'] === 'Y') {
if ($this->withTransaction) {
$this->jf->db->commit($transactionId);
}
return true;
}
return false;
}
}- Job또한 Interface로 정의할 수 있지만, Job의 유형을 하나로만 정의하므로, Class로 정의하였습니다.
초반에 Job을 생성하게 되면, JobListener, ItemReader, ItemProcessor, ItemWriter를 주입받게 됩니다.
run() 메서드에서는
초반에, JobListener 사전처리를 수행합니다.
다음으로는, ItemReader를 통하여 Item목록을 불러오고, Item목록의 갯수만큼, ItemProcessor -> ItemWriter를 연이어 태우게 됩니다.
마지막으로, JobListener로 돌아와서 afterJob() 을 통해서, 마무리 작업을 진행하게 됩니다.
아주 간단한 작업으로 복잡한 배치를 처리하는 것이죠!
여기서 중요한 부분이 있다고 하면, JobParameter에 대한 존재입니다.
JobParameter는 외부에서 Job객체에 run()을 호출할 때 넣어주게 되는데요,
실행할 기준을 전달하는 인자를 넘겨줍니다. 같은 Job이라고 할지라도, 기준에 따라서 수행되는 데이터가 완전히 달라질 수 있기 때문에 반드시 필요한 개념입니다.
그 다음으로는 JobContext라는 개념이 존재합니다. 하나의 Job에서 배치처리할 때는, 캐싱데이터가 반드시 필요하게 됩니다.
캐싱데이터의 역할은 생각보다 매우 큰 역할을 가지게 됩니다.
- 설정 데이터를 미리 메모리에 로드하여, 배치 처리 중 변경되는 설정 값들에 영향을 받지 않도록 도와준다.
- 배치 처리 도중에 DB접근을 최소 접근할 수 있도록 도와준다.
- 배치 모듈간에, 데이터를 공유할 수 있는 유일한 통로이다.
위의 특징으로, JobContext는 배치처리에 있어서, 반드시 필요한 개념이 되어야 합니다.
JobContext가 모듈화된 ItemProcessor, ItemWriter들이 데이터가 동기되어 처리될 수 있게 도와주는 역할을 수행하는 것입니다.
JobListener, ItemReader, ItemProcessor, ItemWriter는 생각보다 간단한 인터페이스로 구성되어 있습니다.
<?php
interface JobListener {
/**
* 작업을 실행하기 전에 사전에 호출하는 메서드.
* @param $jobContext
* @param $param
* @return mixed
*/
public function beforeJob(&$jobContext, $param);
/**
* 작업을 실행한 이후에 호출하는 메서드.
* @param $jobContext
* @param $param
* @return mixed
*/
public function afterJob(&$jobContext, $param);
}
interface ItemReader {
/**
* 데이터를 읽어들여서, 리스트로 반환하는 함수
* @param $jobContext
* @return mixed
*/
public function read(&$jobContext);
}
interface ItemProcessor {
/**
* 데이터를 item 단위로 하나씩 처리하는 함수
* @param $item
* @param $jobContext
* @return mixed
*/
public function process($item, &$jobContext);
}
interface ItemWriter {
/**
* 데이터를 item 단위로 작성하는 함수.
* @param $item
* @param $jobContext
* @return mixed
*/
public function write($item, &$jobContext);
}- 정말 단순한 형태의 Interface로 구성되어 있습니다.
Job은 이런 단순한 Interface의 구현체를 실행하는 역할만 담당하게 되는 것입니다.
단순함에 복잡성을 부여하기
위의 과정은 정말로 단순합니다. 그리고 그 단순함이 정말 핵심이 됩니다.
위의 구조를 흐트리지 않는 선에서, 약간의 유연한 복잡성을 추가해보겠습니다.
ItemProcessor, ItemWriter에 말이죠.
하나의 배치에 하나의 ItemProcessor만 존재한다면, 조건에 따른 다양한 처리, 연속적인 가공처리 진행에 어려움을 겪게 됩니다.
이런 관점에서, 새로운 기능을 가진 ItemProcessor를 도입하려고 합니다.
CompositeItemProcessor
<?php
class CompositeItemProcessor implements ItemProcessor {
private array $itemProcessorList = array();
public function __construct(array $itemProcessorList) {
if ( empty($itemProcessorList) ) {
$this->itemProcessorList = array();
return;
}
$this->itemProcessorList = $itemProcessorList;
}
public function process($item, &$jobContext) {
foreach($this->itemProcessorList AS $itemProcessor) {
$item = $itemProcessor->process($item, $jobContext);
}
return $item;
}
}- CompositeItemProcessor은 ItemProcessor를 직접 구현한 구현체 class 입니다.
- 역할은 여러 개의 ItemProcessor를 담아서, 순차적으로 실행하는 역할을 담당합니다.
- ItemProcessor가 없다면, 그냥 item만 리턴하게 되는 것이구요.
ClassifierItemProcessor
<?php
class ClassifierItemProcessor implements ItemProcessor {
private $classifier;
public function __construct(callable $classifier) {
$this->classifier = $classifier;
}
public function process($item, &$jobContext) {
$itemProcessor = ($this->classifier)($item, $jobContext);
if ( $itemProcessor ) {
$item = $itemProcessor->process($item, $jobContext);
}
return $item;
}
}- ClassfierItemProcessor도 마찬가지로, ItemProcessor의 구현체입니다. 특별한 기능은 별도의 구현체로 만들어 내부에서 또 다른 구현체를 받는 구조인 것이죠.
- 생성자에서, 함수를 입력받습니다. PHP는 함수를 1급 객체로 취급할 수 있습니다. 따라서 함수로 조건을 정의하고 이를 인자로 넣어서 사용하는 수 있습니다.
- 여기서 $classifier 라는 함수는, $item을 통해서, $Item의 상태의 값을 확인하여 어떤 ItemProcessor를 처리할지 결정합니다.
- 이를 사용하게 되면, 배치 Job을 구성할 때, 조건 분기 처리를 통해서, 어떨 때는 어떤 ItemProcessor를 사용하라고 명시할 수 있습니다.
- 물론, 명시된 ItemProcessor가 또 다른 ClassifierItemProcessor가 될 수도 있고, CompositeItemProcessor가 될 수도 있고, 그냥 ItemProcessor를 구현한 로직 구현체가 될 수도 있습니다. 이는 확장성이 무한하다는 것을 의미합니다.
- 그렇게 되면, Job 입장에서는 단 하나의 ItemProcessor 구현체를 호출했지만, 실제로는 이에 종속된 CompositeItemProcessor, ClassifierItemProcessor가 호출되어, 복잡한 조건부 로직을 유연하게 처리할 수 있는 형태가 완성됩니다.
ItemWriter도 같은 방식으로 구현됨을 참고하시면 됩니다.
실제 사용을 위한 예시
// 리스너 선언.
$testjobListener = new TestJobListener();
// ItemReader 선언.
$testItemReader = new TestItemReader();
// ItemProcessor 선언.
$testItemProcessor = new TestItemProcessor();
// ItemWriter 선언.
$testItemWriter = new TestItemWriter();
// Job 생성하기.
$job = new Job(
$testjobListener,
$testItemReader,
$testItemProcessor,
$testItemWriter,
array(
'with_transaction' => 't',
)
);
$job->run(array(
'test_job_id' => 'test-job-id-1',
'job_mode' => 'AUTO',
));- 위의 형태가 가장 기본적으로 사용하는 예제가 될 것입니다.
- 하나의 배치 작업에, 데이터를 읽고, 하나의 ItemProcessor, 하나의 ItemWriter를 거쳐서 배치 작업이 완료되는 단순한 구조의 예제입니다.
- 아주 단순한 목적의 배치이니까, 사실 위의 케이스는 그냥 직접 모듈화 하지 않고, 짜는 것이 더 간단하고, 직관성이 뛰어날 수 있습니다.
하지만, 실제로 실무에서는 복잡한 배치처리가 기다리고 있습니다.
이런 경우에 직접 모듈화 없이 코드를 작성하게 된다면, if/else문이 매우 많아지고, N중첩 형태의 Arrow 코드가 생기고, 중복되는 코드도 발생하고,
코드를 유지보수하기 어려운 그런 형태가 됩니다. 하나가 수정되면, 어디까지 영향이 퍼지는지 감조차도 잡기가 어렵기 때문입니다.
이런 경우에는 구조적인 배치처리를 도입해야 합니다.
구조적인 배치는 ItemProcessor와 ItemWriter를 조건에 따른, 순차적 배치를 구성하면서 시작됩니다.
/**
* 예제: 조건 기반 데이터 처리 파이프라인 구성
*
* 흐름:
* 1. 공통 전처리
* 2. 모드에 따른 처리 분기
* 3. 데이터 상태에 따른 추가 분기
* 4. 후처리
*/
private function buildItemProcessor(): ItemProcessor
{
// 1️⃣ 모든 데이터가 반드시 거치는 공통 전처리
$commonProcessor = new CommonItemProcessor();
// 2️⃣ 특정 모드에서만 실행되는 프로세서
$autoModeProcessor = new AutoModeItemProcessor();
$modeClassifier = new ClassifierItemProcessor(
function ($item, $jobContext) use ($autoModeProcessor) {
$mode = $jobContext['job_parameter']['mode'] ?? 'MANUAL';
switch ($mode) {
case 'AUTO':
return $autoModeProcessor;
default:
return null; // 아무 처리도 하지 않음
}
}
);
// 3️⃣ 후속 처리 단계들
$enrichmentProcessor = new EnrichmentItemProcessor(); // 데이터 보강
$assignmentProcessor = new AssignmentItemProcessor(); // 담당자/카테고리 할당
// 정상 데이터 처리 파이프라인
$normalPipeline = new CompositeItemProcessor([
$modeClassifier,
$enrichmentProcessor,
$assignmentProcessor,
]);
// 예외 데이터 처리 프로세서
$exceptionProcessor = new ExceptionItemProcessor();
$exceptionPipeline = new CompositeItemProcessor([
$exceptionProcessor,
$assignmentProcessor, // 예외 데이터도 기본 할당은 필요
]);
// 4️⃣ 데이터 상태에 따라 분기
$statusClassifier = new ClassifierItemProcessor(
function ($item) use ($normalPipeline, $exceptionPipeline) {
if (!empty($item['is_exception'])) {
return $exceptionPipeline;
}
if (empty($item['is_disabled'])) {
return $normalPipeline;
}
return null; // 처리 제외
}
);
// 최종 파이프라인 구성
return new CompositeItemProcessor([
$commonProcessor,
$statusClassifier,
]);
}
- 위와 같이 다양한 ItemProcessor를 구성하고, 이를 ClassifierItemProcessor(조건부 Item 처리기), CompositeItemProcessor(연결 Item 처리기) 로 합쳐 구성하여,
- 최종적으로는 하나의 ItemProcessor를 만들게 됩니다.
- 매우 복잡한 구조처럼 보일 수는 있으나, 이는 먼저 흐름도(Flow Chart)를 그려보고, 거기에 맞춰 순차적으로 조립해서 구성하기만 하면 됩니다.
구현체 로직 Class 구현하기.
위에서 JobListener, ItemReader, ItemProcessor, ItemWriter 가 조합되어서, 다양한 배치 형태를 만들 수 있음을 알게되었지만,
가장 중요한 실제 비즈니스 로직이 담긴 구현체를 어떻게 만드는 부분이 필요합니다.
사실 이 부분은 자유도가 높아서, 입력과 출력만 맞으면 되는 형태라서, 간단한 구현형태만 소개하겠습니다.
JobListener
<?php
/**
* 예제: Job 실행 전/후에 필요한 작업을 처리하는 Listener
*/
class ExampleJobListener implements JobListener
{
/**
* Job 시작 전에 실행
* - 공통 설정이나 캐시 데이터를 준비해서 jobContext에 저장
*/
public function beforeJob(&$jobContext, $param)
{
// 예: 외부 설정값 또는 기준 데이터 로드
$jobContext['config'] = [
'batch_size' => 100,
'run_mode' => $param['mode'] ?? 'DEFAULT',
];
echo "Job is starting...\n";
}
/**
* Job 종료 후 실행
* - 처리 결과를 기록하거나 후속 작업 수행
*/
public function afterJob(&$jobContext, $param)
{
echo "Job has finished.\n";
// 예: 처리 결과가 있으면 저장
if (!empty($jobContext['processed_count'])) {
$result = [
'processed_count' => $jobContext['processed_count'],
'finished_at' => date('Y-m-d H:i:s'),
];
// 실제 환경이라면 DB 저장 또는 로그 기록
print_r($result);
}
}
}
- JobListener의 beforeJob() 에서는 데이터의 설정값을 조회해서, $JobContext에 Caching하는 과정을 담당.
- afterJob()은 결과에 따른 추가 데이터 저장 처리를 담당.
ItemReader
<?php
/**
* 예제: 데이터베이스에서 처리할 데이터를 읽어오는 ItemReader
*/
class ExampleItemReader implements ItemReader
{
private $db;
public function __construct()
{
// DB 연결 (예시)
$this->db = Database::getInstance();
}
public function read(&$jobContext): array
{
echo "Reading items for batch...\n";
// Job 파라미터 활용
$days = $jobContext['job_parameter']['days'] ?? 7;
$query = "
SELECT *
FROM items
WHERE is_active = 1
AND created_at >= DATE_SUB(NOW(), INTERVAL {$days} DAY)
ORDER BY created_at
";
$items = $this->db->fetch($query);
echo "Loaded " . count($items) . " items.\n";
return $items;
}
}
- ItemReader는 read메서드에, 배치처리할 대상 목록을 조회하고, 해당 목록을 return 하는 역할.
ItemProcessor
/**
* 예제: 데이터를 검증하고 상태를 추가하는 ItemProcessor
*/
class ExampleItemProcessor implements ItemProcessor
{
public function process($item, &$jobContext)
{
echo "Processing item: {$item['id']}\n";
// 1️⃣ 차단 대상인지 확인
if ($this->isBlocked($item)) {
$item['is_disabled'] = true;
return $item;
}
// 2️⃣ 데이터가 유효한지 검증
if (!$this->isValid($item)) {
$item['is_disabled'] = true;
return $item;
}
// 3️⃣ 중복 데이터 확인
if ($this->isDuplicate($item)) {
$item['is_duplicate'] = true;
return $item;
}
return $item;
}
private function isBlocked($item): bool
{
// 예: 특정 조건이면 차단
return false;
}
private function isValid($item): bool
{
// 예: 필수 값 존재 여부 체크
return !empty($item['value']);
}
private function isDuplicate($item): bool
{
// 예: 이미 처리된 데이터인지 확인
return false;
}
}
- ItemProcessor는 데이터의 내용을 확인하고, 조건에 따라 $item을 가공처리하는 역할
ItemWriter
<?php
/**
* 예제: 처리된 데이터를 외부 서비스로 전송하는 ItemWriter
*/
class ExampleItemWriter implements ItemWriter
{
private $notificationService;
public function __construct()
{
// 예: 알림 서비스 또는 API 클라이언트
$this->notificationService = new NotificationService();
}
public function write($item, &$jobContext)
{
echo "Writing item: {$item['id']}\n";
$messages = $item['messages'] ?? [];
foreach ($messages as $message) {
$this->notificationService->send($message);
}
return $item;
}
}
- ItemWriter는 $item에 가공된 데이터를 실제로 DB/외부서비스에 기록하는 역할 수행.
주의사항!
코드를 보시면 알겠지만, 구조화는 되어있으나, 사실 내부적으로 어떤 로직을 수행하든 개발자의 몫 입니다.
그러므로, Processor과정에서 DB를 조작하는 경우가 발생할 가능성도 있습니다. 이런 형태의 구현은 유지보수에 좋지 않습니다. 지양하는 편이 좋습니다.
ItemReader SELECT만 처리, ItemProcessor에서는 SELECT와 $item에 대한 가공만 처리!, ItemWriter에서는 INSERT, UPDATE, DELETE만 처리!
이런 형태로 명확히 구분해야, 나중에 리스크가 있는 작업을 쉽게 파악할 수 있습니다.
마무리
저는 배치처리를 하면서 느끼는 것은,
꽤 단순한 로직 몇 가지로, 모듈화를 쉽게 할 수 있고, 복잡한 로직을 구성할 수 있다는 사실이 신기하였습니다.
저는 현실세계도 마찬가지라고 생각합니다.
어떤 것을 수행하는데, 복잡함이 먼저 따른다고 하면, 근본적으로 잘못되었을 가능성이 높다고 생각합니다.
단순한 것들이 모여서 복잡함을 구성해야지, 처음부터 기반자체가 복잡하면, 소프트웨어는 알 수 없는 버그와 오류가 많이 발생할 수 밖에 없습니다.
복잡함은 단순함으로부터 확장되어 나타나게 됩니다.
해당 게시글을 공유할 수 있어 좋은 시간이 되었습니다.
답글 남기기