0002 Reactive Programming 4

이번에는 Reactive Stream을 이용해서 실질적으로 현업에서 사용할 만한 코드를 작성해보겠다. 

 

 요구사항 

 

   - Oracle DB , OJdbc, DBCP2

   - org.reactivesreams

   - ExecutorService

 

  A, B라는 Oracle DB User에 대해서 TB_USER로 부터 USER_ID를 가져와 출력하라는 요구 사항을 접수 받았다. 하지만 이 DB User의 수는 상황에 따라서 증가하거나 줄어들 수 있으며 그에 따라 고객이 빠른 응답을 받을 수 있도록 구성이 필요하다고 하자. 이와 같은 요구사항을 ReactiveStream(Java8)을 이용해서 구현해본다. 

---

1번 : Main 진입점 

 

 DataSource를 생성한뒤 각각의 Command 객체 생성시 전달하여 DB Select시에 사용하려고 한다. 

 해당 부분은 Controller 를 통해서 전달받은 파라미터 및 구분자에 의해서 동적으로 변경될 수 있는 부분이라고 가정한다. 

즉 commandHadlerList의 배열이 요청 갯수에 따라서 증가하거나 감소할 수 있다. 

public static void main(String[] args) {

        BasicDataSource dataSource = new BasicDataSource();
        dataSource.setDriverClassName("oracle.jdbc.driver.OracleDriver");
        dataSource.setUrl("jdbc:oracle:thin:@***********:****/****");
        dataSource.setUsername("*****");
        dataSource.setPassword("*****");

        BasicDataSource dataSource2 = new BasicDataSource();
        dataSource2.setDriverClassName("oracle.jdbc.driver.OracleDriver");
        dataSource2.setUrl("jdbc:oracle:thin:@***********:****/****");
        dataSource2.setUsername("*****");
        dataSource2.setPassword("*****");

        CommandHandler commandHandler1 = new CommandHandler(1);

        commandHandler1.add(new SelectDBCommand(dataSource));
        commandHandler1.add(new SelectDBCommand(dataSource2));

        List<CommandHandler> commandHandlerList = new ArrayList<>();

        commandHandlerList.add(commandHandler1);

        Publisher<CommandHandler> publisher = new SamplePublisher(commandHandlerList);

        Subscriber<CommandHandler> subscriber = new SampleSubscriber(commandHandlerList.size());

        publisher.subscribe(subscriber);
        
}

2번 : Publisher 

 

 Observer 패턴의  Observable과 같은 역할을 하며, Subscriber를 구독할 수 있게 설정하여 실행시 Publihser에서 Subscriber로 onNext, onComplete, onError를 처리할 수 있다. 

public class SamplePublisher implements Publisher<CommandHandler> {

    private final List<CommandHandler> commandHandlerList;

    public SamplePublisher(List<CommandHandler> commandHandlerList){
        this.commandHandlerList = commandHandlerList;
    }

    @Override
    public void subscribe(Subscriber<? super CommandHandler> subscriber) {
        subscriber.onSubscribe(new Subscription() {
            @Override
            public void request(long n) {
                commandHandlerList.forEach(subscriber::onNext);

                // 데이터 전송 처리 완료
                subscriber.onComplete();
            }

            @Override
            public void cancel() {

            }
        });
    }
}

3번 : Subcriber - Publisher 내부에서 호출한 각기 함수(onNext, onComplete, onError)에 맞게 진행된다. 

 

 SampleSubscriber가 생성되는 시점에 만들어진 ExectorService에 대해서  Java의 Future를 활용하여 onComplete 시점에 ExecutorService를 Shutdown 시킬 수 있게 처리하였다. 

public class SampleSubscriber implements Subscriber<CommandHandler> {

    private final ExecutorService executorService;

    List<Future<?>> futures = new ArrayList<>();

    public SampleSubscriber(int executeCount){
        this.executorService = Executors.newFixedThreadPool(executeCount);
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        subscription.request(1);
    }

    @Override
    public void onNext(CommandHandler commandHandler) {
        futures.add(executorService.submit(commandHandler::StartAPICall));
    }

    @Override
    public void onError(Throwable t) {

    }

    @Override
    public void onComplete() {
        for(Future<?> future : futures) {
            try {
                future.get();
            }catch(Exception e){
                e.printStackTrace();
            }
        }
        executorService.shutdown();
    }
}

4번 : Command Handler 

 

3번 항목과 처리되는 방식은 거의 유사하며, Command Handler에서 각각의 Command를 ExecutorService를 이용하여  실행시키는 구조이다. 

public class CommandHandler {

    private final List<Command> commandList = new ArrayList<>();
    List<Future<?>> futures = new ArrayList<>();

    private final ExecutorService executorService;

    public CommandHandler(int threadCount){
        executorService = Executors.newFixedThreadPool(threadCount);
    }

    public void StartAPICall() {

        commandList.forEach(command -> {
            futures.add(executorService.submit(command::execute, "Success"));
        });

        for(Future<?> future : futures) {
            try {
                future.get();
            }catch(Exception e){
                e.printStackTrace();
            }
        }

        executorService.shutdown();
    }

    public synchronized void add(Command command) {
        commandList.add(command);
    }
}

5번 : Command 

public interface Command {
    public void execute();
}

6번 : Command 구현 - SelectDBCommand 

 

 실질적으로 데이터를 조회하고, 이를 용도에 맞게 처리하는 Command의 구현체이다. SelectDBCommand 객체가 

생성되는 시점에 필요한 정보를 모두 합성 방식을 이용하여 미리 준비하고, execute가 호출 되는 시점에 활용한다. 

아래와 같은 방식으로 구성했을 때 ThreadSafe하게 구현되었다고 할 수 있다. 

public class SelectDBCommand implements Command {

    private final DataSource dataSource;

    public SelectDBCommand(DataSource dataSource){
        this.dataSource =  dataSource;
    }

    @Override
    public void execute() {

        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);

        List<User> list =  jdbcTemplate.query("SELECT USER_ID FROM *********", new CustomMapper());

        System.out.println("START" + Thread.currentThread().getName());
        list.stream().forEach(item -> {
            System.out.print(item.getUSER_ID() + ",");
        });
        System.out.println("END" + Thread.currentThread().getName());
    }
}

7번 : CustomMapper - JDBC Rowmapper 

 

JdbcTemplate의 RowMapper를 구현하여 필요에 맞게 값을 바인딩 처리한다. 

public class CustomMapper implements RowMapper<User> {
    @Override
    public User mapRow(ResultSet resultSet, int i) throws SQLException {
        User user = new User();
        user.setUSER_ID(resultSet.getString("USER_ID"));
        return user;
    }
}