대용량 처리 테스트 구현 프로젝트 - 2 (코드 작성)

📂 Redis 관련 파일

1. RedisConfig.java

역할	RedisTemplate을 설정하여 Redis 서버와 상호작용
설명	Redis와의 연결을 설정하고, 데이터를 문자열로 직렬화 및 역직렬화하는 데 필요한 설정을 포함

package com.realTime.Insight.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(connectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer()); // 키를 문자열로 직렬화
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer()); // 값을 JSON으로 직렬화
        return redisTemplate;
    }

    @Bean
    public RedisMessageListenerContainer redisContainer(RedisConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listenerAdapter, new PatternTopic("webSocketTopic")); // 특정 주제의 메시지를 수신
        return container;
    }

}

2. DataService.java

역할	Redis를 이용해 데이터를 저장하고 조회하는 서비스 클래스
설명	Redis에 데이터를 추가하고 조회하는 로직 포함

package com.realTime.Insight;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;

@Service
public class DataService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private ValueOperations<String, Object> valueOperations;

    @PostConstruct
    public void init() {
        valueOperations = redisTemplate.opsForValue();
    }

    public List getData() {
        return redisTemplate.opsForHash().values("TimeSeriesData").stream()
                .map(data -> (Data) data)
                .collect(Collectors.toList());
    }

    public void addData(Data data) {
        String key = "webSocketTopic:" + System.currentTimeMillis();
        valueOperations.set(key, data);  // 단일 값 저장
        redisTemplate.opsForHash().put("TimeSeriesData", key, data);  // 해시 값 저장
    }

    public List getAllData() {
        List dataList = new ArrayList<>();
        for (String key : redisTemplate.keys("webSocketTopic:*")) {
            dataList.add(valueOperations.get(key));
        }
        return dataList;
    }
}

---

📂 Kafka 관련 파일

1. KafkaConsumerConfig.java

역할	Kafka Consumer 설정을 담당
설명	Kafka의 Consumer를 설정하고, 메시지를 처리할 Listener를 설정

package com.realTime.Insight.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> config = new HashMap<>();

        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

2. KafkaProducerConfig.java

역할	Kafka Producer 설정을 담당
설명	Kafka의 Producer를 설정하고, KafkaTemplate을 제공

package com.realTime.Insight.config;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;

@EnableKafka
@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

3. KafkaProducerService.java

역할	Kafka를 통해 메시지를 전송하는 서비스 클래스
설명	Kafka에 메시지를 보내는 로직을 포함, sendMessage 메서드는 지정된 토픽으로 메시지를 전송

package com.realTime.Insight;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * Kafka로 메시지를 전송하는 메서드
     * @param topic 메시지를 전송할 Kafka 토픽
     * @param message 전송할 메시지
     */
    public void sendMessage(String topic, Object message) {
        kafkaTemplate.send(topic, message);
    }
}

4. KafkaConsumerService.java

역할	Kafka로부터 메시지를 소비하는 서비스 클래스
설명	Kafka 메시지를 소비하고, 이를 Redis 및 WebSocket으로 전송하는 로직 포함

package com.realTime.Insight;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @KafkaListener(topics = "webSocketTopic", groupId = "group_id")
    public void consume(String message) {
        redisTemplate.opsForList().rightPush("real_time_data", message);
        System.out.println("message => " + message);
        messagingTemplate.convertAndSend("/topic/data", message);
        System.out.println("Consumed message: " + message);
    }
}

---

💻 Front 파일 (React)

DataDisplay.js 파일을 아래와 같이 생성합니다.

import React, { useEffect, useState } from 'react';
import axios from 'axios';
import SockJS from 'sockjs-client';
import { Stomp } from '@stomp/stompjs';

const DataDisplay = () => {
    const [data, setData] = useState([]);

    useEffect(() => {
        // Fetch initial data
        axios.get('http://localhost:8080/api/data')
            .then(response => {
                setData(response.data);
            })
            .catch(error => {
                console.error('There was an error fetching the data!', error);
            });

        // Setup WebSocket connection
        const socket = new SockJS('http://localhost:8080/ws');
        const stompClient = Stomp.over(socket);

        stompClient.connect({}, frame => {
            console.log('Connected: ' + frame);
            stompClient.subscribe('/topic/data', message => {
                const newData = JSON.parse(message.body);
                setData(prevData => [...prevData, newData]);
            });
        });

        return () => {
            if (stompClient) {
                stompClient.disconnect();
            }
        };
    }, []);

    return (Data from Backend
{data.map((item) => (
ID: {item.id}Title: {item.title}
Description: {item.description}
Timestamp: {item.timestamp}
))}
    );
};

export default DataDisplay;

 

중요 코드는 위와 같이 작성하고, 나머지 코드는 아래 GitHub를 참고합니다.

GitHub - sonhoil/Largevolume: 대용량 처리 시스템 테스트 환경 구현 프로젝트

GitHub - sonhoil/Largevolume: 대용량 처리 시스템 테스트 환경 구현 프로젝트