[java] Apache ActiveMQ를 이용한 실시간 데이터 처리

23 Nov 2023

java

소개

Apache ActiveMQ는 자바 기반의 오픈 소스 메시지 브로커입니다. 이를 통해 실시간으로 데이터를 처리하고, 메시지를 송수신할 수 있습니다. 이 기술은 대규모 분산 시스템에서 많은 양의 데이터를 신속하게 처리할 수 있는 강력한 도구입니다.

설치 및 설정

ActiveMQ 홈페이지에서 ActiveMQ를 다운로드하고 설치합니다.
설치된 ActiveMQ를 시작합니다. (./activemq start)
ActiveMQ 관리자 화면에 접속하여 로그인합니다. (기본 URL: http://localhost:8161/admin)
필요에 따라 브로커 설정을 수정하고, 토픽(Topic)과 큐(Queue)을 생성합니다.

데이터 송신

ActiveMQ를 이용해 실시간으로 데이터를 송신하는 방법은 다음과 같습니다.

import javax.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;

public class DataProducer {
    public static void main(String[] args) {
        try {
            // ActiveMQ 연결 설정
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            Connection connection = connectionFactory.createConnection();
            connection.start();

            // 메시지를 보낼 프로듀서 생성
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createTopic("myTopic");
            MessageProducer producer = session.createProducer(destination);

            // 데이터를 생성하여 메시지로 변환하여 보냄
            String data = "Hello, ActiveMQ!";
            TextMessage message = session.createTextMessage(data);
            producer.send(message);

            // 작업 종료
            connection.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

데이터 수신

ActiveMQ를 이용해 실시간으로 데이터를 수신하는 방법은 다음과 같습니다.

import javax.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;

public class DataConsumer {
    public static void main(String[] args) {
        try {
            // ActiveMQ 연결 설정
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            Connection connection = connectionFactory.createConnection();
            connection.start();

            // 메시지를 받을 컨슈머 생성
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createTopic("myTopic"); // 송신자와 동일한 토픽을 구독
            MessageConsumer consumer = session.createConsumer(destination);

            // 메시지를 수신하여 데이터 출력
            consumer.setMessageListener(message -> {
                if (message instanceof TextMessage) {
                    try {
                        String data = ((TextMessage) message).getText();
                        System.out.println("Received data: " + data);

                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

            // 데이터 수신 대기
            Thread.sleep(5000);

            // 작업 종료
            connection.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

결론

Apache ActiveMQ를 사용하여 실시간으로 데이터를 처리하는 방법에 대해 알아보았습니다. 이를 통해 여러 시스템간에 데이터를 신속하게 전달하고 처리할 수 있습니다. ActiveMQ의 강력한 기능과 다양한 설정 옵션을 활용하여 실시간 데이터 처리를 더욱 효과적으로 구현할 수 있습니다.

더 많은 정보를 원하시면 Apache ActiveMQ 공식 문서를 참고하시기 바랍니다.