[Dining-together] web socket && stomp && spring boot && Redis 로 rest 채팅서비스 (micro service)
3
Dining-together
목록 보기
21/25
web socket && stomp && spring boot && Redis 로 rest 채팅서비스 (micro service)
websocket이란
WebSocket은 서버와 클라이언트 간에 Socket Connection을 유지해서 언제든 양방향 통신 또는 데이터 전송이 가능하도록 하는 기술이다.
Real-time web application구현을 위해 널리 사용되어지고 있다. (SNS어플리케이션, LoL같은 멀티플레이어 게임, 구글 Doc, 증권거래, 화상채팅 등)
http를 두고 소켓으로 통신하는 이유는?
웹어플리케이션에서 기존의 서버와 클라이언트 간의 통신은 대부분 HTTP를 통해 이루어 졌으며 HTTP는 Request/response기반의 Stateless protocol이다.
즉, 서버와 클라이언트 간의 Socket connection같은 영구적인 연결이 되어있지 않고 클라이언트 쪽에서 필요할때 Request를 할때만 서버가 Response를 하는 방식으로 통신이 진행되는 한방향 통신이다. 이럴경우 서버쪽 데이터가 업데이트 되더라도 클라이언트 쪽에는 화면은 Refresh하지 않는한 변경된 데이터가 업데이트 되지 않는 문제가 발생한다. 이런 문제는 일반적은 웹어플리케이션에서는 기존의 있던 임시방편인 Long polling이라던가 Ajax를 사용해도 어느정도 해결이 가능하지만 데이터의 빠른 업데이트가 아주 중요한 요소 중에 하나인 어플리케이션에서는 실시간 업데이트가 아주 중요하기 때문에 Web Socket이 아주 중요한 기술로 사용되고 있다.
Web Socket은 Stateful protocol이기 때문에 클라이언트와 한 번 연결이 되면 계속 같은 라인을 사용해서 통신하기 때문에 HTTP 사용시 필요없이 발생되는 HTTP와 TCP연결 트래픽을 피할 수 있다. 마지막으로 Web Socket은 HTTP와 같은 포트(80)을 사용하기에 기업용 어플리케이션에 적용할 때 방화벽은 재설정 하지 않아도 되는 장점이 있다.
Stomp
Stomp는 메시징 전송을 효율적으로 하기 위해 나온 프로토콜이며 기본적으로 pub/sub 구조로 되어있어 메시지를 발송하고, 메시지를 받아 처리하는 부분이 확실히 정해져 있기 때문에 개발하는 입장에서 명확하게 인지하고 개발할 수 있는 이점이 있습니다. 또한 Stomp를 이용하면 통신 메시지의 헤더에 값을 세팅할 수 있어 헤더 값을 기반으로 통신 시 인증 처리를 구현하는 것도 가능.
pub/sub란 메시지를 공급하는 주체와 소비하는 주체를 분리하여 제공하는 메시징 방법. 기본적인 콘셉트를 예를 들면 우체통(topic)이 있으면 집배원(publisher)이 신문을 우체동에 배달하는 액션이 있고, 우체통에 신문이 배달되는 것을 기다렸다가 빼서 보는 구독자(subscriber)의 액션이 있다.
- 채팅방 생성 - topic 생성(방id)
- 채팅방 입장 - topic 구독 ( subscribe roomId)
- 메시지를 주고 받는다 - pub 메시지 발송 sub 메시지 받음 -> messagemapping
Redis 이용하는 이유?
- 서버를 재시작 할때마다 채팅방 정보 리셋 -> Db 저장이 필요
- 채팅서버가 여러대면 서버간 채팅방을 공유할 수 없음
websocket과 stomp에 pub/sub를 이용해 채팅방을 구현했다. 그런데 이 구조는 pub/sub가 발생한 서버 내에서만 메시지를 주고 받는것이 가능하다. 즉, 구독 대상인 채팅방(topic)이 생성된 서버 안에서만 유요하므로 다른 서버에서 접속해도 보이지 않는다. 여렇기 때문에 구독 대상 채팅방이 여러 서버에 접근할 수 있도록 개선해야한다. 공통으로 사용하는 pub/sub 시스템을 구축하여 모든 서버가 해당 시스템을 통해 메시지를 주고받을 수 있도록 구현해야한다.
코드
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.4.7-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>kr.or.dining_together</groupId>
<artifactId>chat</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>chat</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>11</java.version>
<spring-cloud.version>2020.0.2</spring-cloud.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
<dependency>
<groupId>org.modelmapper</groupId>
<artifactId>modelmapper</artifactId>
<version>2.3.8</version>
</dependency>
<!-- Swagger Api Documentation&UI -->
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.9.2</version>
</dependency>
<!-- Feign Client -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
<dependency>
<groupId>it.ozimov</groupId>
<artifactId>embedded-redis</artifactId>
<version>0.7.2</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
<version>2.5.4</version>
</dependency>
<!-- redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.6.7</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>2.7.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind -->
<!-- json support -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/com.h2database/h2 -->
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>jcenter-snapshots</id>
<name>jcenter</name>
<url>http://oss.jfrog.org/artifactory/oss-snapshot-local/</url>
</repository>
</repositories>
</project>
config
RedisConfig
RedisConnectionFactory를 통해 redisconnection을 생성해 redis를 연결한다.
RedisTemplate - Redis를 직접적으로 사용한다. RedisConnecttion에서 넘겨준 byte값을 객체 직렬화 한다. 스레드세이프하며 여러개의 인스턴스에서 재사용될수 있다.
@RequiredArgsConstructor
@Configuration
public class RedisConfig {
/**
* redis pub/sub 메시지를 처리하는 listener 설정
*/
@Bean
public RedisMessageListenerContainer redisMessageListener(RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
return container;
}
/**
* 어플리케이션에서 사용할 redisTemplate 설정
*/
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(connectionFactory);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<>(String.class));
return redisTemplate;
}
}WebSocketConfig
WebSocketMessageBrokerConfigurer 구현한다.
Websocket api를 바로 사용하지 않고 STOMP를 통해서 설정한다.
-
configureMessageBroker : 메시지 브로커에 관련된 설정을 한다
-
registerStompEndpoints : SockJs Fallback을 이용해 노출할 STOMP endpoint를 설정한다.
메시지 발행하는 prefix는 /pub로 시작하도록 설정하고 메시지를 구독하는 요청의 prefix는 /sub 시작하도록 설정했다. stomp websocket의 연결 endpoint는 /chatting 으로 설정했다.
@RequiredArgsConstructor
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
// @Override
// public void configureClientInboundChannel(ChannelRegistration registration) {
// registration.interceptors(stompHandler);
// }
@Override
public void configureMessageBroker(MessageBrokerRegistry config) {
config.enableSimpleBroker("/sub");
config.setApplicationDestinationPrefixes("/pub");
}
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/chatting").setAllowedOrigins("*")
.withSockJS();
}
}entity
@Getter
@Builder
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PROTECTED)
@Entity
public class ChatMessage {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
private MessageType type; // 메시지 타입
private String sender; // 메시지 보낸사람
private String message; // 메시지
@ManyToOne(fetch = FetchType.LAZY)
@JoinColumn(name = "chatroom_id")
private ChatRoom chatRoom;
public static ChatMessage createChatMessage(ChatRoom chatRoom, String sender, String message,MessageType type) {
ChatMessage chatMessage= ChatMessage.builder()
.chatRoom(chatRoom)
.sender(sender)
.message(message)
.type(type)
.build();
return chatMessage;
}
public void setSender(String sender){
this.sender=sender;
}
public void setMessage(String message){
this.message=message;
}
// 메시지 타입 : 입장, 퇴장, 채팅
public enum MessageType {
ENTER, QUIT, TALK
}
}package kr.or.dining_together.chat.model;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import javax.persistence.CascadeType;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.OneToMany;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
@Entity
public class ChatRoom implements Serializable {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Column(name = "chatroom_id")
private String id;
private static final long serialVersionUID = 6494678977089006639L;
private String name;
private UserIdDto customer;
private UserIdDto store;
@OneToMany(mappedBy = "chatRoom", cascade = CascadeType.ALL)
private List<ChatMessage> chatMessages = new ArrayList<>();
public static ChatRoom create(String name,UserIdDto customer,UserIdDto store) {
ChatRoom chatRoom = new ChatRoom();
chatRoom.id = UUID.randomUUID().toString();
chatRoom.name = name;
chatRoom.customer=customer;
chatRoom.store=store;
return chatRoom;
}
public void addChatMessages(ChatMessage chatMessage) {
this.chatMessages.add(chatMessage);
}
}
repository
public interface ChatRoomRepository extends JpaRepository<ChatRoom,String> {
List<ChatRoom> findChatRoomsByCustomer(UserIdDto customer);
List<ChatRoom> findChatRoomsByStore(UserIdDto store);
}service
입장 퇴장 처리 / 방생성 입장 처리를 위해 redis
private HashOperations<String, String, ChatRoom> opsHashChatRoom;
private HashOperations<String, String, String> hashOpsEnterInfo;
를 만들어 value key값을 넣어주었다.
db에 chatroom과 chatmessage를 따로 추가적으로 저장했다.
package kr.or.dining_together.chat.service;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Service;
import kr.or.dining_together.chat.model.ChatMessage;
import kr.or.dining_together.chat.model.ChatRoom;
import kr.or.dining_together.chat.model.UserIdDto;
import kr.or.dining_together.chat.pubsub.RedisSubscriber;
import kr.or.dining_together.chat.repository.ChatRoomRepository;
import lombok.RequiredArgsConstructor;
@RequiredArgsConstructor
@Service
public class ChatService {
// 채팅방(topic)에 발행되는 메시지를 처리할 Listner
private final RedisMessageListenerContainer redisMessageListener;
// 구독 처리 서비스
private final RedisSubscriber redisSubscriber;
// Redis
private static final String CHAT_ROOMS = "CHAT_ROOM";
public static final String ENTER_INFO = "ENTER_INFO"; // 채팅룸에 입장한 클라이언트의 sessionId와 채팅룸 id를 맵핑한 정보 저장
private final RedisTemplate<String, Object> redisTemplate;
private HashOperations<String, String, ChatRoom> opsHashChatRoom;
// 채팅방의 대화 메시지를 발행하기 위한 redis topic 정보. 서버별로 채팅방에 매치되는 topic정보를 Map에 넣어 roomId로 찾을수 있도록 한다.
private Map<String, ChannelTopic> topics;
private HashOperations<String, String, String> hashOpsEnterInfo;
@PostConstruct
private void init() {
opsHashChatRoom = redisTemplate.opsForHash();
hashOpsEnterInfo=redisTemplate.opsForHash();
topics = new HashMap<>();
}
private final ChatRoomRepository chatRoomRepository;
public List<ChatRoom> findAllRoom() {
return chatRoomRepository.findAll();
}
public ChatRoom findRoomById(String id) {
ChatRoom chatRoom=(ChatRoom)chatRoomRepository.findById(id).orElseThrow();
return chatRoom;
}
/**
* 채팅방 생성 : 서버간 채팅방 공유를 위해 redis hash에 저장한다.
*/
public ChatRoom createChatRoom(UserIdDto customer, UserIdDto store) {
String name=customer.getName()+"와 "+store.getName();
ChatRoom chatRoom = ChatRoom.create(name,customer,store);
opsHashChatRoom.put(CHAT_ROOMS, chatRoom.getId(), chatRoom);
chatRoomRepository.save(chatRoom);
return chatRoom;
}
/**
* 채팅방 입장 : redis에 topic을 만들고 pub/sub 통신을 하기 위해 리스너를 설정한다.
*/
public void enterChatRoom(String roomId) {
ChannelTopic topic = topics.get(roomId);
if (topic == null)
topic = new ChannelTopic(roomId);
redisMessageListener.addMessageListener(redisSubscriber, topic);
topics.put(roomId, topic);
}
public ChannelTopic getTopic(String roomId) {
return topics.get(roomId);
}
public List<ChatRoom> getCustomerEnterRooms(UserIdDto customer){
return chatRoomRepository.findChatRoomsByCustomer(customer);
}
public List<ChatRoom> getStoreEnterRooms(UserIdDto store){
return chatRoomRepository.findChatRoomsByStore(store);
}
public void deleteById(ChatRoom chatRoom){
chatRoomRepository.delete(chatRoom);
}
/**
* destination정보에서 roomId 추출
*/
public String getRoomId(String destination) {
int lastIndex = destination.lastIndexOf('/');
if (lastIndex != -1)
return destination.substring(lastIndex + 1);
else
return "";
}
// 유저가 입장한 채팅방ID와 유저 세션ID 맵핑 정보 저장
public void setUserEnterInfo(String sessionId, String roomId) {
hashOpsEnterInfo.put(ENTER_INFO, sessionId, roomId);
}
// 유저 세션으로 입장해 있는 채팅방 ID 조회
public String getUserEnterRoomId(String sessionId) {
return hashOpsEnterInfo.get(ENTER_INFO, sessionId);
}
// 유저 세션정보와 맵핑된 채팅방ID 삭제
public void removeUserEnterInfo(String sessionId) {
hashOpsEnterInfo.delete(ENTER_INFO, sessionId);
}
}pub sub
@RequiredArgsConstructor
@Service
public class RedisPublisher {
private final RedisTemplate<String, Object> redisTemplate;
public void publish(ChannelTopic topic, ChatMessage message) {
redisTemplate.convertAndSend(topic.getTopic(), message);
}
}@Slf4j
@RequiredArgsConstructor
@Service
public class RedisSubscriber implements MessageListener{
private final ObjectMapper objectMapper;
private final RedisTemplate redisTemplate;
private final SimpMessageSendingOperations messagingTemplate;
/**
* Redis에서 메시지가 발행(publish)되면 대기하고 있던 onMessage가 해당 메시지를 받아 처리한다.
*/
@Override
public void onMessage(Message message, byte[] pattern) {
try {
// redis에서 발행된 데이터를 받아 deserialize
String publishMessage = (String) redisTemplate.getStringSerializer().deserialize(message.getBody());
// ChatMessage 객채로 맵핑
ChatMessage roomMessage = objectMapper.readValue(publishMessage, ChatMessage.class);
// Websocket 구독자에게 채팅 메시지 Send
messagingTemplate.convertAndSend("/sub/chat/room/" + roomMessage.getChatRoom().getId(), roomMessage);
} catch (Exception e) {
log.error(e.getMessage());
}
}
}
controller
chattingcontroller
@MessageMapping("url") : "url"으로 들어오는 메시지 매핑할때 사용하는 애노테이션
"/pub/chat/message" 으로 들어오는 message를 ChatMessage으로 바인딩하여 실행
@RequiredArgsConstructor
@Slf4j
@RestController
@CrossOrigin
public class ChattingController {
private final RedisPublisher redisPublisher;
private final UserServiceClient userServiceClient;
private final ChatService chatService;
/**
* websocket "/pub/chat/message"로 들어오는 메시징을 처리한다.
*/
@ApiOperation(value = "채팅방 메시지", notes = "메시지")
@MessageMapping("/chat/message")
public void message(ChatMessageDto message, @RequestHeader("X-AUTH-TOKEN") String xAuthToken) {
UserIdDto user = userServiceClient.getUserId(xAuthToken);
// 로그인 회원 정보로 대화명 설정
ChatRoom chatRoom=chatService.findRoomById(message.getRoomId());
ChatMessage message1=ChatMessage.createChatMessage(chatRoom, user.getName(), message.getMessage(), message.getType());
// 채팅방 입장시에는 대화명과 메시지를 자동으로 세팅한다.
log.info("채팅 메시지");
if (ChatMessage.MessageType.ENTER.equals(message1.getType())) {
message1.setSender("[알림]");
message1.setMessage(user.getName() + "님이 입장하셨습니다.");
}else if(ChatMessage.MessageType.QUIT.equals(message1.getType())){
message1.setSender("[알림]");
message1.setMessage(user.getName() + "님이 퇴장하셨습니다.");
chatService.deleteById(message1.getChatRoom());
}
chatRoom.addChatMessages(message1);
// Websocket에 발행된 메시지를 redis로 발행(publish)
redisPublisher.publish(chatService.getTopic(message.getRoomId()), message1);
}
}@RequiredArgsConstructor
@RestController
@RequestMapping("/chat")
public class ChatRoomController {
private final ChatService chatService;
private final ResponseService responseService;
private final UserServiceClient userServiceClient;
@ApiOperation(value = "room 전체 조회", notes = "채팅 룸 전체를 조회한다.")
@GetMapping("/rooms")
public ListResult<ChatRoom> rooms() {
return responseService.getListResult(chatService.findAllRoom());
}
@ApiOperation(value = "채팅방 개설", notes = "채팅방을 개설한다.")
@PostMapping("/room")
public SingleResult<ChatRoom> createRoom(@RequestHeader("X-AUTH-TOKEN") String xAuthToken,@RequestBody UserIdDto store) {
UserIdDto customer = userServiceClient.getUserId(xAuthToken);
return responseService.getSingleResult(chatService.createChatRoom(customer,store));
}
@ApiOperation(value = "방 정보 보기", notes = "방 정보")
@GetMapping("/room/{roomId}")
public SingleResult<ChatRoom> roomInfo(@PathVariable String roomId) {
return responseService.getSingleResult(chatService.findRoomById(roomId));
}
@ApiOperation(value = "customer 별 방 조회")
@GetMapping("/customer")
public ListResult<ChatRoom> getRoomsByCustomer(@RequestHeader("X-AUTH-TOKEN") String xAuthToken){
UserIdDto customer=userServiceClient.getUserId(xAuthToken);
return responseService.getListResult(chatService.getCustomerEnterRooms(customer));
}
@ApiOperation(value = "store 별 방 조회")
@GetMapping("/store")
public ListResult<ChatRoom> getRoomsByStore(@RequestHeader("X-AUTH-TOKEN") String xAuthToken){
UserIdDto store=userServiceClient.getUserId(xAuthToken);
return responseService.getListResult(chatService.getCustomerEnterRooms(store));
}
}참고문서
https://duckdevelope.tistory.com/19
https://daddyprogrammer.org/post/4691/spring-websocket-chatting-server-stomp-server/
