---

1. 메세징(Messaging)
   • 10.1. JMS
     • 10.1.1. ActiveMQ 지원(ActiveMQ Support)
     • 10.1.2. ActiveMQ 아르테미스 지원(ActiveMQ Artemis Support)
     • 10.1.3. JNDI 커넥션팩토리 사용(Using a JNDI ConnectionFactory)
     • 10.1.4. 메세지 전송(Sending a Message)
     • 10.1.5. 메세지 받기(Receiving a Message)
   • 10.2. AMQP
     • 10.2.1. 래빗MQ 지원(RabbitMQ Support)
     • 10.2.2. 메세지 전송(Sending a Message)
     • 10.2.3. 스트림으로 메시지 전송(Sending a Message To A Stream)
     • 10.2.4. 메세지 받기(Receiving a Message)
   • 10.3. 아파치 카프카 지원(Apache Kafka Support)
     • 10.3.1. 메세지 전송(Sending a Message)
     • 10.3.2. 메세지 받기(Receiving a Message)
     • 10.3.3. 카프카 스트림(Kafka Streams)
     • 10.3.4. 카프카 추가 프로퍼티스(Additional Kafka Properties)
     • 10.3.5. 임베디드 카프카를 사용한 테스트(Testing with Embedded Kafka)
   • 10.4. R소켓(RSocket)
     • 10.4.1. R소켓 전략 자동 구성(RSocket Strategies Auto-configuration)
     • 10.4.2. R소켓 서버 자동 구성(RSocket server Auto-configuration)
     • 10.4.3. 스프링 메세징 R소켓 지원(Spring Messaging RSocket support)
     • 10.4.4. R소켓리퀘스터를 사용하여 R소켓 서비스 호출(Calling RSocket Services with RSocketRequester)
   • 10.5. 스프링 인테그레이션(Spring Integration)
   • 10.6. 웹소켓(WebSockets)
   • 10.7. 다음에 읽을 내용(What to Read Next)

---

10. 메세징(Messaging)

스프링 프레임워크는 JmsTemplate을 사용하는 JMS API의 단순한 사용부터 메시지를 비동기적으로 수신하는 인프라에 이르기까지 메시징 시스템과의 통합을 위한 광범위한 지원을 제공한다. 스프링 AMQP는 Advanced Message Queuing Protocol과 유사한 기능을 제공한다. 스프링 부트는 래빗템플릿(RabbitTemplate) 및 RabbitMQ에 대한 자동 구성 옵션도 제공한다. 스프링 웹소켓(Spring WebSocket)에는 기본적으로 STOMP 메시징 지원이 포함되어 있으며 스프링 부트는 스타터 자동 구성을 통해 이를 지원한다. 스프링 부트는 아파치 카프카도 지원한다.

10.1. JMS

jakarta.jms.ConnectionFactory 인터페이스는 JMS 브로커와 상호 작용하기 위해 jakarta.jms.Connection을 생성하는 표준 방법을 제공한다. 스프링이 JMS와 작동하려면 커넥션팩토리(ConnectionFactory)가 필요하지만 일반적으로 이를 직접 사용할 필요는 없으며 대신 더 높은 레벨의 메시징 추상화에 의존할 수 있다. (자세한 내용은 스프링 프레임워크 레퍼런스 문서의 관련 절을 참고하자.) 또한 스프링 부트는 메시지를 보내고 받는 데 필요한 인프라를 자동 구성한다.

10.1.1. ActiveMQ 지원(ActiveMQ Support)

클래스패스에서 ActiveMQ를 사용할 수 있으면 스프링 부트는 커넥션팩토리(ConnectionFactory)를 구성할 수 있다.

spring-boot-starter-activemq를 사용하는 경우 JMS와 통합하기 위한 스프링 인프라와 마찬가지로 ActiveMQ 인스턴스에 연결하는 데 필요한 의존성이 제공된다.

ActiveMQ 구성은 spring.activemq.*의 외부 구성 프로퍼티스에 의해 제어된다. 기본적으로 ActiveMQ는 TCP 전송을 사용하도록 자동 구성되어 기본적으로 tcp://localhost:61616에 연결된다. 다음 예제에서는 기본 브로커 URL을 변경하는 방법을 보여준다.

프로퍼티스(Properties)

spring.activemq.broker-url=tcp://192.168.1.210:9876
spring.activemq.user=admin
spring.activemq.password=secret

Yaml

spring:
  activemq:
    broker-url: "tcp://192.168.1.210:9876"
    user: "admin"
    password: "secret"

기본적으로, 캐싱커넥션팩토리(CachingConnectionFactory)는 spring.jms.*의 외부 구성 프로퍼티스로 제어할 수 있는 합리적인 설정으로 기본 커넥션팩토리(ConnectionFactory)를 래핑한다.

프로퍼티스(Properties)

spring.jms.cache.session-cache-size=5

Yaml

spring: 
  jms:
    cache:
      session-cache-size: 5

기본 풀링을 사용하려는 경우, 다음 예제와 같이 org.messaginghub:pooled-jms에 의존성을 추가하고 그에 따라 Jms풀커넥션팩토리(JmsPoolConnectionFactory)를 구성하여 그렇게 할 수 있다.

프로퍼티스(Properties)

spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=50

Yaml

spring:
  activemq:
    pool:
      enabled: true
      max-connections: 50

지원되는 옵션에 대한 자세한 내용은 액티브MQ프로퍼티스(ActiveMQProperties)를 참고하자. 또한 고급 커스텀를 위해 액티브MQ커넥션팩토리커스터마이저(ActiveMQConnectionFactoryCustomizer)를 구현하는 임의 개수의 빈을 등록할 수도 있다.

기본적으로, ActiveMQ는 대상이 아직 존재하지 않는 경우 제공된 대상을 이름을 확인 후 생성한다.

10.1.2. ActiveMQ 아르테미스 지원(ActiveMQ Artemis Support)

스프링 부트는 클래스패스에서 ActiveMQ 아르테미스(Artemis)를 사용할 수 있음을 감지하면 커넥션팩토리(ConnectionFactory)를 자동으로 구성할 수 있다. mode 프로퍼티가 명시적으로 설정되지 않은 경우, 브로커가 있으면 포함된 브로커가 자동으로 시작되고 구성된다. 지원되는 모드는 임베디드(임베디드 브로커가 필요하며 클래스패스에서 브로커를 사용할 수 없는 경우 오류가 발생해야 함을 명시하기 위해) 및 네이티브(네티 전송 프로토콜을 사용하여 브로커에 연결하기 위해) 모드다. 후자가 구성되면 스프링 부트는 기본 설정으로 로컬 시스템에서 실행되는 브로커에 연결하는 커넥션팩토리(ConnectionFactory)를 구성한다.

spring-boot-starter-artemis를 사용하는 경우 기존 ActiveMQ 아르테미스 인스턴스에 연결하는 데 필요한 의존성은 물론 JMS와 통합하기 위한 스프링 인프라도 제공된다. org.apache.activemq:artemis-jakarta-server를 애플리케이션에 추가하면 임베디드 모드를 사용할 수 있다.

ActiveMQ 아르테미스 구성은 spring.artemis.*의 외부 구성 프로퍼티스에 의해 제어된다. 다음 예제는 application.properties에서 선언할 수 있는 프로퍼티를 보여준다.

프로퍼티스(Properties)

spring.artemis.mode=native
spring.artemis.broker-url=tcp://192.168.1.210:9876
spring.artemis.user=admin
spring.artemis.password=secret

Yaml

spring:
  artemis:
    mode: native
    broker-url: "tcp://192.168.1.210:9876"
    user: "admin"
    password: "secret"

브로커를 내장할 때, 지속성(persistence) 활성화 여부를 선택하고 사용 가능한 대상을 나열할 수 있다. 이는 쉼표로 구분된 목록으로 지정하여 기본 옵션으로 생성하거나 org.apache.activemq.artemis.jms.server.config.JMSQueueConfiguration 또는 org.apache.activemq 타입의 빈을 정의할 수 있다. artemis.jms.server.config.TopicConfiguration - 각각 고급 큐 및 토픽 구성용이다.

기본적으로, 캐싱커넥션팩토리(CachingConnectionFactory)는 spring.jms.*의 외부 구성 프로퍼티스로 제어할 수 있는 설정이며 기본 커넥션팩토리(ConnectionFactory)를 래핑한다.

프로퍼티스(Properties)

spring.jms.cache.session-cache-size=5

Yaml

spring: 
  jms:
    cache:
      session-cache-size: 5

네이티브 풀링을 사용하려는 경우, 다음 예제와 같이 org.messaginghub:pooled-jms에 대한 의존성을 추가하고 그에 따라 Jms풀커넥션팩토리(JmsPoolConnectionFactory)를 구성할 수 있다.

프로퍼티스(Properties)

spring.artemis.pool.enabled=true
spring.artemis.pool.max-connections=50

Yaml

spring:
  artemis:
    pool:
      enabled: true
      max-connections: 50

지원되는 추가 옵션은 아르테미스프로퍼티스(ArtemisProperties)를 참고하자.

JNDI 조회는 포함되지 않으며, 대상은 아르테미스(Artemis) 구성의 애트리뷰트명 또는 구성으로 제공된 이름을 사용하여 이름을 확인한다.

10.1.3. JNDI 커넥션팩토리 사용(Using a JNDI ConnectionFactory)

애플리케이션 서버에서 애플리케이션을 실행 중인 경우 스프링 부트는 JNDI를 사용하여 JMS 커넥션팩토리(ConnectionFactory)를 찾는다. 기본적으로 java:/JmsXA 및 java:/XAConnectionFactory 위치가 확인된다. 다음 예제와 같이 대체 위치를 지정해야 하는 경우 spring.jms.jndi-name 프로퍼티를 사용할 수 있다.

프로퍼티스(Properties)

spring.jms.jndi-name=java:/MyConnectionFactory

Yaml

spring: 
  jms:
    jndi-name: "java:/MyConnectionFactory"

10.1.4. 메세지 전송(Sending a Message)

스프링의 Jms템플릿(JmsTemplate)은 자동으로 구성되며 다음 예제와 같이 빈에 자동 연결할 수 있다.

자바

import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

@Component
public class MyBean {
    private final JmsTemplate jmsTemplate;

    public MyBean(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void someMethod() {
        this.jmsTemplate.convertAndSend("hello");
    } 
}

코틀린

import org.springframework.jms.core.JmsTemplate
import org.springframework.stereotype.Component

@Component
class MyBean(private val jmsTemplate: JmsTemplate) {
    fun someMethod() {
        jmsTemplate.convertAndSend("hello")
    } 
}

Jms메세징템플릿(JmsMessagingTemplate)도 비슷한 방식으로 주입할 수 있다. 데스티네이션리졸버(DestinationResolver) 또는 메세지컨버터(MessageConverter) 빈이 정의되면 자동으로 구성된 Jms템플릿(JmsTemplate)에 자동으로 연결된다.

10.1.5. 메세지 받기(Receiving a Message)

JMS 인프라가 있는 경우 모든 빈에 @JmsListener 어노테이션을 추가하여 리스너 엔드포인트를 생성할 수 있다. Jms리스너컨테이너팩토리(JmsListenerContainerFactory)가 정의되지 않은 경우, 기본 항목이 자동으로 구성된다. 데스티네이션리졸버(DestinationResolver), 메세지컨버터(MessageConverter) 또는 jakarta.jms.ExceptionListener 빈이 정의된 경우 기본 팩토리와 자동으로 연관된다.

기본적으로, 기본 팩토리는 트랜잭션이다. Jta트랜젝션매니저(JtaTransactionManager)가 있는 인프라에서 실행하는 경우 기본적으로 리스너 컨테이너에 연결된다. 그렇지 않은 경우 sessionTransacted 플래그가 활성화된다. 후자의 시나리오에서는 리스너 메서드(또는 그 델리게이트(delegate))에 @Transactional을 추가하여 로컬 데이터 저장소 트랜잭션을 수신 메시지 처리에 연결할 수 있다. 이렇게 하면 로컬 트랜잭션이 완료되면 들어오는 메시지가 승인된다. 여기에는 동일한 JMS 세션에서 수행된 응답 메시지 전송도 포함된다.

다음 컴포넌트는 someQueue 대상에 리스너 엔드포인트를 생성한다.

자바

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {
    @JmsListener(destination = "someQueue")
    public void processMessage(String content) {
        // ... 
    }
}

코틀린

import org.springframework.jms.annotation.JmsListener
import org.springframework.stereotype.Component

@Component
class MyBean {
    @JmsListener(destination = "someQueue")
    fun processMessage(content: String?) {
        // ... 
    }
}

자세한 내용은 @EnableJms의 자바독(Javadoc)을 참고하자.

더 많은 Jms리스너컨테이너팩토리(JmsListenerContainerFactory) 인스턴스를 생성해야 하거나 기본값을 오버라이드하려는 경우 스프링 부트는 자동 구성된 것과 동일한 설정으로 디폴트Jms리스너컨테이너팩토리(DefaultJmsListenerContainerFactory)를 초기화하는 데 사용할 수 있는 디폴트Jms리스너컨테이너팩토리컨피규어러(DefaultJmsListenerContainerFactoryConfigurer)를 제공한다.

다음 예제에서는 특정 메세지컨버터(MessageConverter)를 사용하는 다른 팩토리를 노출한다.

자바

import jakarta.jms.ConnectionFactory;
import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;

@Configuration(proxyBeanMethods = false)
public class MyJmsConfiguration {
    @Bean
    public DefaultJmsListenerContainerFactorymyFactory(DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        ConnectionFactory connectionFactory = getCustomConnectionFactory();
        configurer.configure(factory, connectionFactory);
        factory.setMessageConverter(new MyMessageConverter());
        return factory;
    }

    private ConnectionFactory getCustomConnectionFactory() {
        return ...
    }
}

코틀린

import jakarta.jms.ConnectionFactory
import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.jms.config.DefaultJmsListenerContainerFactory

@Configuration(proxyBeanMethods = false)
class MyJmsConfiguration {
    @Bean
    fun myFactory(configurer: DefaultJmsListenerContainerFactoryConfigurer): DefaultJmsListenerContainerFactory {
        val factory = DefaultJmsListenerContainerFactory()
        val connectionFactory = getCustomConnectionFactory()
        configurer.configure(factory, connectionFactory)
        factory.setMessageConverter(MyMessageConverter())
        return factory
    }

    fun getCustomConnectionFactory() : ConnectionFactory? {
        return ...
    } 
}

그런 다음 다음과 같이 @JmsListener 어노테이션이 달린 메서드에서 팩토리를 사용할 수 있다.

자바

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {
    @JmsListener(destination = "someQueue", containerFactory = "myFactory")
    public void processMessage(String content) {
        // ... 
    }
}

코틀린

import org.springframework.jms.annotation.JmsListener
import org.springframework.stereotype.Component

@Component
class MyBean {
    @JmsListener(destination = "someQueue", containerFactory = "myFactory")
    fun processMessage(content: String?) {
        // ... 
    }
}

10.2. AMQP

AMQP(Advanced Message Queuing Protocol)는 메시지 지향 미들웨어를 위한 플랫폼 중립적, 와이어 레벨(wire-level) 프로토콜이다. 스프링 AMQP 프로젝트는 AMQP 기반 메시징 솔루션 개발에 코어 스프링 개념을 적용한다. 스프링 부트는 spring-boot-starter-amqp “스타터(Starter)”를 포함하여 레빗MQ를 통해 AMQP로 작업할 수 있는 여러 가지 편의를 제공한다.

10.2.1. 래빗MQ 지원(RabbitMQ Support)

래빗MQ는 AMQP 프로토콜을 기반으로 하는 가볍고 안정적이며 확장 가능하고 이식 가능한 메시지 브로커다. 스프링은 AMQP 프로토콜을 통해 통신하기 위해 래빗MQ를 사용한다.

래빗MQ 구성은 spring.rabbitmq.*의 외부 구성 프로퍼티스에 의해 제어된. 다음 예제는 application.properties의 선언을 보여준다.

프로퍼티스(Properties)

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=secret

Yaml

spring:
  rabbitmq:
    host: "localhost"
    port: 5672
    username: "admin"
    password: "secret"

또는, 다음과 같이 addresses 애트리뷰트를 사용하여 동일한 커넥션을 구성할 수도 있다.

프로퍼티스(Properties)

spring.rabbitmq.addresses=amqp://admin:secret@localhost

Yaml

spring:
  rabbitmq:
    addresses: "amqp://admin:secret@localhost"

이런 방식으로 addresses를 지정하면 호스트 및 포트 프로퍼티스가 무시된다. addresses가 amqps 프로토콜을 사용하는 경우 SSL 지원이 자동으로 활성화된다.

지원되는 프로퍼팃 구성 옵션에 대한 자세한 내용은 래빗프로퍼티스(RabbitProperties)를 참고하자. 스프링 AMQP에서 사용되는 래빗MQ 커네션팩토리(ConnectionFactory)의 하위 레벨 디테일을 구성하려면 커넥션팩토리커스터마이저(ConnectionFactoryCustomizer) 빈을 정의하자.

커넥션네임스트래티지(ConnectionNameStrategy) 빈이 컨텍스트에 존재하는 경우 자동 구성된 캐싱커넥션팩토리(CachingConnectionFactory)에 의해 생성된 커넥션명을 지정하는 데 자동으로 사용된다.

애플리케이션 전체에 걸쳐 래빗템플릿(RabbitTemplate)에 대한 추가 커스텀을 수행하려면 래빗템플릿커스터마이저(RabbitTemplateCustomizer) 빈을 사용하자.

See Understanding AMQP, the protocol used by RabbitMQ for more details.

10.2.2. 메세지 전송(Sending a Message)

스프링의 Amqp템플릿(AmqpTemplate) 및 Ampq어드민(AmqpAdmin)은 자동 구성되며, 다음 예제와 같이 이를 사용자 고유의 빈에 직접 오토와이어드(autowire)할 수 있다.

자바

import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Component;

@Component
public class MyBean {
    private final AmqpAdmin amqpAdmin;
    
    private final AmqpTemplate amqpTemplate;

    public MyBean(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) {
        this.amqpAdmin = amqpAdmin;
        this.amqpTemplate = amqpTemplate;
    }

    public void someMethod() {
        this.amqpAdmin.getQueueInfo("someQueue");
    }

    public void someOtherMethod() {
        this.amqpTemplate.convertAndSend("hello");
    } 
}

코틀린

import org.springframework.amqp.core.AmqpAdmin
import org.springframework.amqp.core.AmqpTemplate
import org.springframework.stereotype.Component

@Component
class MyBean(private val amqpAdmin: AmqpAdmin, private val amqpTemplate: AmqpTemplate) {
    fun someMethod() {
        amqpAdmin.getQueueInfo("someQueue")
    }

    fun someOtherMethod() {
        amqpTemplate.convertAndSend("hello")
    } 
}

래빗메세징템플릿(RabbitMessagingTemplate)도 비슷한 방식으로 주입할 수 있다. 메세지컨버터(MessageConverter) 빈이 정의되면, 자동으로 구성된 Ampq템플릿(AmqpTemplate에) 자동으로 연결된다.

필요한 경우, 빈으로 정의된 org.springframework.amqp.core.Queue는 래빗MQ 인스턴스에서 해당 큐을 선언하는 데 자동으로 사용된다.

작업을 재시도하려면, AmqpTemplate에서 재시도(retry)를 활성화하면 된다(예: 브로커 연결이 끊어진 경우).

프로퍼티스(Properties)

spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=2s

Yaml

spring:
  rabbitmq:
    template:
      retry:
        enabled: true
        initial-interval: "2s"

재시도는 기본적으로 비활성화되어 있다. 래빗리트라이템플릿커스터마이저(RabbitRetryTemplateCustomizer) 빈을 선언하여 프로그래밍 방식으로 리트라이템플릿(RetryTemplate)을 커스텀할 수도 있다.

더 많은 래빗템플릿(RabbitTemplate) 인스턴스를 생성해야 하거나 기본값을 오버라이드하려는 경우 스프링 부트는 자동 구성에서 사용되는 팩토리와 동일한 설정으로 래빗템플릿(RabbitTemplate)을 초기화하는 데 사용할 수 있는 래빗템플릿컨피규어러(RabbitTemplateConfigurer) 빈을 제공한다.

10.2.3. 스트림으로 메시지 전송(Sending a Message To A Stream)

특정 스트림에 메시지를 보내려면 다음 예제와 같이 스트림명을 지정하자.

프로퍼티스(Properties)

spring.rabbitmq.stream.name=my-stream

Yaml

spring:
  rabbitmq:
    stream:
      name: "my-stream"

메세지컨버터(MessageConverter), 스트림메세지컨버터(StreamMessageConverter) 또는 프로듀서커스터마이저(ProducerCustomizer) 빈이 정의된 경우 자동 구성된 래빗스트림템플릿(RabbitStreamTemplate)에 자동으로 연결된다.

10.2.4. 메세지 받기(Receiving a Message)

래빗 인프라가 있는 경우 모든 빈에 @RabbitListener 어노테이션을 추가하여 리스너 엔드포인트를 생성할 수 있다. 래빗리스너컨테이너팩토리(RabbitListenerContainerFactory)가 정의되지 않은 경우 기본 심플래빗리스너컨테이너팩토리(SimpleRabbitListenerContainerFactory)가 자동으로 구성되며 spring.rabbitmq.listener.type 프로퍼티를 사용하여 직접 컨테이너로 전환할 수 있다. 메세지컨버터(MessageConverter) 또는 메세지리커버러(MessageRecoverer) 빈이 정의되면 자동으로 기본 팩토리와 연결된다.

다음 샘플 컴포넌트는 someQueue 큐에 리스너 엔드포인트를 생성한다.

자바

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {
    @RabbitListener(queues = "someQueue")
    public void processMessage(String content) {
        // ... 
    }
}

코틀린

import org.springframework.amqp.rabbit.annotation.RabbitListener
import org.springframework.stereotype.Component

@Component
class MyBean {
    @RabbitListener(queues = ["someQueue"])
    fun processMessage(content: String?) {
        // ... 
    }
}

자세한 내용은 @EnableRabbit의 자바독(Javadoc)을 참고하자

많은 래빗리스너컨테이너팩토리(RabbitListenerContainerFactory) 인스턴스를 생성해야 하거나 기본값을 오버라이드하려는 경우 스프링 부트는 자동 구성에서 사용되는 팩토리와 동일한 설정으로 심플래빗리스너컨테이너팩토리(SimpleRabbitListenerContainerFactory) 및 다이렉트래빗리스너컨테이너팩토리(DirectRabbitListenerContainerFactory)를 초기화하는 데 사용할 수 있는 심플래빗리스너컨테이너팩토리컨피규어러(SimpleRabbitListenerContainerFactoryConfigurer) 및 다이렉트래빗리스너컨테이너팩토리컨피규어러(DirectRabbitListenerContainerFactoryConfigurer)를 제공한다.

어떤 컨테이너 타입을 선택했는지는 중요하지 않다. 이 두 개의 빈은 자동 구성에 의해 노출된다.

예를 들어, 다음 구성 클래스는 특정 메세지컨버터(MessageConverter)를 사용하는 다른 팩터리를 노출한다.

자바

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration(proxyBeanMethods = false)
public class MyRabbitConfiguration {
    @Bean
    public SimpleRabbitListenerContainerFactorymyFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        ConnectionFactory connectionFactory = getCustomConnectionFactory();
        configurer.configure(factory, connectionFactory);
        factory.setMessageConverter(new MyMessageConverter());
        return factory;
    }

    private ConnectionFactory getCustomConnectionFactory() {
        return ...
    } 
}

코틀린

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory
import org.springframework.amqp.rabbit.connection.ConnectionFactory
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration

@Configuration(proxyBeanMethods = false)
class MyRabbitConfiguration {
    @Bean
    fun myFactory(configurer: SimpleRabbitListenerContainerFactoryConfigurer): SimpleRabbitListenerContainerFactory {
        val factory = SimpleRabbitListenerContainerFactory()
        val connectionFactory = getCustomConnectionFactory()
        configurer.configure(factory, connectionFactory)
        factory.setMessageConverter(MyMessageConverter())
        return factory
    }
    
    fun getCustomConnectionFactory() : ConnectionFactory? {
        return ...
    } 
}

그런 다음 다음과 같이 @RabbitListener 어노테이션이 달린 메서드에서 팩토리를 사용할 수 있다.

자바

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {
    @RabbitListener(queues = "someQueue", containerFactory = "myFactory")
    public void processMessage(String content) {
        // ... 
    }
}

코틀린

import org.springframework.amqp.rabbit.annotation.RabbitListener
import org.springframework.stereotype.Component

@Component
class MyBean {
    @RabbitListener(queues = ["someQueue"], containerFactory = "myFactory")
    fun processMessage(content: String?) {
        // ... 
    }
}

리스너가 예외를 발생시키는 상황을 처리하기 위해 재시도를 활성화할 수 있다. 기본적으로 리젝트엔돈트리큐리커버러(RejectAndDontRequeueRecoverer)가 사용되지만 고유한 메세지리커버러(MessageRecoverer)를 정의할 수 있다. 재시도 횟수가 모두 소모되면 메시지가 거부되고 삭제되거나 브로커가 그렇게 하도록 구성된 경우 배달 못한 편지 교환으로 라우팅된다. 기본적으로 재시도는 비활성화되어 있다. 래빗리트라이템플릿커스터마이저(RabbitRetryTemplateCustomizer) 빈을 선언하여 프로그래밍 방식으로 리트라이템플릿(RetryTemplate)을 사용자 정의할 수도 있다.

기본적으로 재시도가 비활성화되어 있고 리스너에서 예외가 발생하면 전달이 무기한 재시도된다. 이 동작은 두 가지 방법으로 수정할 수 있다. defaultRequeueRejected 프로퍼티를 false로 설정하여 재전송을 시도하지 않거나 Amqp리젝트엔돈트리큐익셉션(AmqpRejectAndDontRequeueException)을 발생시켜 메시지가 거부되어야 한다는 신호를 보낸다. 후자는 재시도가 활성화되고 최대 배달 시도 횟수에 도달했을 때 사용되는 메커니즘이다.

10.3. 아파치카프카 지원(Apache Kafka Support)

아파치 카프카는 spring-kafka 프로젝트에서 자동 구성이 지원된다.

카프카 구성은 spring.kafka.*의 외부 구성 프로퍼티스에 의해 제어된다. 다음 예제는 application.properties에서 선언할 수 있는 프로퍼티를 보여준다.

프로퍼티스(Properties)

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=myGroup

Yaml

spring:
  kafka:
    bootstrap-servers: "localhost:9092"
    consumer:
      group-id: "myGroup"

시작 시 토픽(topic)를 생성하려면 뉴토큰(NewTopic) 타입의 빈을 추가하자. 토픽이 이미 존재하는 경우 빈은 무시된다.

지원되는 추가 옵션은 카프카프로퍼티스(KafkaProperties)를 참고하자.

10.3.1. 메세지 전송(Sending a Message)

스프링의 카프카템플릿(KafkaTemplate)은 자동 구성되며 다음 예제와 같이 빈에서 직접 자동 연결할 수 있다.

자바

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class MyBean {
    private final KafkaTemplate<String, String> kafkaTemplate;

    public MyBean(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void someMethod() {
        this.kafkaTemplate.send("someTopic", "Hello");
    } 
}

코틀린

import org.springframework.kafka.core.KafkaTemplate
import org.springframework.stereotype.Component

@Component
class MyBean(private val kafkaTemplate: KafkaTemplate<String, String>) {
    fun someMethod() {
        kafkaTemplate.send("someTopic", "Hello")
    } 
}

spring.kafka.producer.transaction-id-prefix 프로퍼티가 정의되면 카프카트랜젝션매니저(KafkaTransactionManager)가 자동으로 구성된다. 또한 레코드메세지컨버터(RecordMessageConverter) 빈이 정의되면 자동으로 구성된 카프카템플릿(KafkaTemplate)에 자동 연결된다.

10.3.2. Receiving a Message

아파치 카프카 인프라가 있는 경우 모든 빈에 @KafkaListener 어노테이션을 추가하면 리스너 엔드포인트를 생성할 수 있다. 카프카리스너컨테이너팩토리(KafkaListenerContainerFactory)가 정의되지 않은 경우 spring.kafka.listener.*에 정의된 키를 사용하여 기본 항목이 자동으로 구성된다.

다음 컴포넌트는 someTopic 토픽에 리스너 엔드포인트를 생성한다.

자바

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {
    @KafkaListener(topics = "someTopic")
    public void processMessage(String content) {
        // ... 
    }
}

코틀린

import org.springframework.kafka.annotation.KafkaListener
import org.springframework.stereotype.Component

@Component
class MyBean {
    @KafkaListener(topics = ["someTopic"])
    fun processMessage(content: String?) {
        // ... 
    }
}

카프카트랜젝션매니저(KafkaTransactionManager) 빈이 정의되면 자동으로 컨테이너 팩토리에 연결된다. 마찬가지로 레코드필터스트래티지(RecordFilterStrategy), 커먼에러핸들러(CommonErrorHandler), 에프터롤백프로세서(AfterRollbackProcessor) 또는 컨슈머어웨어리벨런스리스너(ConsumerAwareRebalanceListener) 빈이 정의되면 자동으로 기본 팩토리에 연결된다.

리스너 타입에 따라 레코드메세지컨버터(RecordMessageConverter) 또는 배치메세지컨버터(BatchMessageConverter) 빈이 기본 팩토리에 연결된다. 배치 리스너에 레코드메세지컨버터(RecordMessageConverter) 빈만 있는 경우 배치메세지컨버터(BatchMessageConverter)에 래핑된다.

커스텀 체인드카프카트랜젝션매니저(ChainedKafkaTransactionManager)는 일반적으로 자동 구성된 카프카트랜젝션매니저(KafkaTransactionManager) 빈을 참조하므로 @Primary로 표시되어야 한다.

10.3.3. 카프카 스트림(Kafka Streams)

아파치 카프카용 스프링은 스트림빌더(StreamsBuilder) 객체를 생성하고 해당 스트림의 생명주기을 관리하기 위한 팩토리 빈을 제공한다. Kafka-streams가 클래스패스에 있고 카프카 스트림이 @EnableKafkaStreams 어노테이션으로 활성화되어 있는 한 스프링 부트는 필수 카프카스트림즈컨피규레이션(KafkaStreamsConfiguration) 빈을 자동 구성한다.

카프카 스트림을 활성화한다는 것은 애플리케이션 ID와 부트스트랩 서버를 설정해야 함을 의미한다. 전자는 spring.kafka.streams.application-id를 사용하여 구성할 수 있으며, 설정되지 않은 경우 기본적으로 spring.application.name으로 설정된다. 후자는 전역적으로 설정하거나 스트림에 대해서만 특별히 오버라이드할 수 있다.

전용 프로퍼티스를 사용하면 여러 가지 프로퍼티스 추가로 사용할 수 있다. 다른 임의의 카프카 프로퍼티스는 spring.kafka.streams.properties 네임스페이스를 사용하여 설정할 수 있다. 자세한 내용은 추가 카프카 프로퍼티스을 참고하자.

팩토리 빈을 사용하려면 다음 예제와 같이 스트림빌더(StreamsBuilder)를 @Bean에 연결하자.

자바

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.support.serializer.JsonSerde;

@Configuration(proxyBeanMethods = false)
@EnableKafkaStreams
public class MyKafkaStreamsConfiguration {
    @Bean
    public KStream<Integer, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<Integer, String> stream = streamsBuilder.stream("ks1In");
        stream.map(this::uppercaseValue).to("ks1Out", Produced.with(Serdes.Integer(), new JsonSerde<>()));
        return stream;
    }

    private KeyValue<Integer, String> uppercaseValue(Integer key, String value) {
        return new KeyValue<>(key, value.toUpperCase());
    }
}

코틀린

import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.streams.KeyValue
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.kstream.KStream
import org.apache.kafka.streams.kstream.Produced
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.annotation.EnableKafkaStreams
import org.springframework.kafka.support.serializer.JsonSerde

@Configuration(proxyBeanMethods = false)
@EnableKafkaStreams
class MyKafkaStreamsConfiguration {
    @Bean
    fun kStream(streamsBuilder: StreamsBuilder): KStream<Int, String> {
        val stream = streamsBuilder.stream<Int, String>("ks1In")
        stream.map(this::uppercaseValue).to("ks1Out", Produced.with(Serdes.Integer(), JsonSerde()))
        return stream
    }

    private fun uppercaseValue(key: Int, value: String): KeyValue<Int?, String?> {
        return KeyValue(key, value.uppercase())
    } 
}

기본적으로 스트림빌더(StreamBuilder) 객체가 관리하는 스트림은 자동으로 시작된다. spring.kafka.streams.auto-startup 프로퍼티를 사용하여 이 동작을 커스텀할 수 있다.

10.3.4. 카프카 추가 프로퍼티스(Additional Kafka Properties)

자동 구성에서 지원되는 프로퍼티스는 부록의 “통합 프로퍼티스(Integration Properties)” 장에 나와 있다. 대부분의 경우, 이러한 프로퍼티스(하이픈 또는 camelCase)는 아파치 카프카로 구분된 프로퍼티스에 직접 매핑된다. 자세한 내용은 아파치 카프카 문서를 참고하자.

이러한 프로퍼티스 중 처음 몇 개는 모든 컴포넌트(생산자(producers), 소비자(consumers), 관리자(admins) 및 스트림(streams))에 적용되지만 다른 값을 사용하려는 경우 컴포넌트 레벨에서 지정할 수 있다. 아파치 카프카는 중요도가 HIGH, MEDIUM 또는 LOW인 프로퍼티스를 지정한다. 스프링 부트는 자동 구성은 중요도가 HIGH인 모든 프로퍼티스, 일부 선택된 MEDIUM 및 LOW 프로퍼티스, 기본값이 없는 모든 프로퍼티스를 지원한다.

카프카에서 지원하는 프로퍼티스 중 일부만 카프카프로퍼티스(KafkaProperties) 클래스를 통해 직접 사용할 수 있다. 직접 지원되지 않는 추가 프로퍼티스를 사용하여 생산자 또는 소비자를 구성하려면 다음 프로퍼티스를 사용하자.

프로퍼티스(Properties)

spring.kafka.properties[prop.one]=first
spring.kafka.admin.properties[prop.two]=second
spring.kafka.consumer.properties[prop.three]=third
spring.kafka.producer.properties[prop.four]=fourth
spring.kafka.streams.properties[prop.five]=fifth

Yaml

spring:
  kafka:
    properties:
      "[prop.one]": "first"
    admin:
      properties:
        "[prop.two]": "second"
    consumer:
      properties:
        "[prop.three]": "third"
    producer:
      properties:
        "[prop.four]": "fourth"
    streams:
      properties:
        "[prop.five]": "fifth"

이는 공통 prop.one 카프카 프로퍼티를 첫 번째(생산자, 소비자 및 관리자에게 적용)로 설정하고, prop.two admin 프로퍼티를 두 번째로, prop.three consumer 프로퍼티를 세 번째로, prop.four producer 프로퍼티를 네 번째로, prop.five streams 프로퍼티를 다섯 번째로 설정한다.

다음과 같이 스프링 카프카 Json디시리얼라이저(JsonDeserializer)를 구성할 수도 있다.

프로퍼티스(Properties)

spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties[spring.json.value.default.type]=com.example.Invoice
spring.kafka.consumer.properties[spring.json.trusted.packages]=com.example.main,com.example.another

Yaml

 spring:
    kafka:
      consumer:
        value-deserializer:
          "org.springframework.kafka.support.serializer.JsonDeserializer"
        properties:
          "[spring.json.value.default.type]": "com.example.Invoice"
          "[spring.json.trusted.packages]": "com.example.main,com.example.another"

마찬가지로, 헤더에 타입 정보를 보내는 Json시리얼라이저(JsonSerializer)의 기본 동작을 비활성화할 수 있다.

프로퍼티스(Properties)

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.properties[spring.json.add.type.headers]=false

Yaml

spring:
  kafka:
    producer:
      value-serializer: "org.springframework.kafka.support.serializer.JsonSerializer"
      properties:
        "[spring.json.add.type.headers]": false

이러한 방식으로 설정된 프로퍼티스는 스프링 부트가 명시적으로 지원하는 모든 구성 항목을 오버라이드한다.

10.3.5. 임베디드 카프카를 사용한 테스트(Testing with Embedded Kafka)

아파치 카프카용 스프링은 임베디드 아파치 카프카 브로커를 사용하여 프로젝트를 테스트하는 편리한 방법을 제공한다. 이 기능을 사용하려면 spring-kafka-test 모듈에서 @EmbeddedKafka로 테스트 클래스에 어노테이션을 달면된다. 자세한 내용은 스프링 아파치 카프카 레퍼런스 매뉴얼을 참고하자.

앞서 언급한 임베디드 아파치 카프카 브로커에서 스프링 부트 자동 구성이 작동하도록 하려면 임베디드 브로커 주소(EmbeddedKafkaBroker로 채워짐)에 대한 시스템 프로퍼티를 아파치 카프카용 스프링 부트 구성 프로퍼티로 다시 매핑해야 한다. 이를 수행하는 방법에는 여러 가지가 있다.

• 임베디드 브로커 주소를 테스트 클래스의 spring.kafka.bootstrap-servers에 매핑하는 시스템 프로퍼티스를 제공한다.

자바

static {
      System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers");
}

코틀린

init {
      System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers")
}

• @EmbeddedKafka 어노테이션에 프로퍼티명을 구성한다.

자바

import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;

@SpringBootTest
@EmbeddedKafka(topics = "someTopic", bootstrapServersProperty =
"spring.kafka.bootstrap-servers")
class MyTest {
    // ... 
}

코틀린

import org.springframework.boot.test.context.SpringBootTest
import org.springframework.kafka.test.context.EmbeddedKafka

@SpringBootTest
@EmbeddedKafka(topics = ["someTopic"], bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class MyTest {
    // ... 
}

• 구성 속성에서 자리 표시자(placeholder)를 사용한다.

프로퍼티스(Properties)

spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}

Yaml

spring:
  kafka:
    bootstrap-servers: "${spring.embedded.kafka.brokers}"

10.4. R소켓(RSocket)

R소켓(RSocket)은 바이트 스트림 전송에 사용되는 바이너리 프로토콜이다. 단일 연결로 전달되는 비동기 메시지를 통해 대칭 상호 작용 모델(symmetric interaction models)을 활성화한다.

스프링 프레임워크의 스프링 메시징 모듈은 클라이언트와 서버 측 모두에서 R소켓(RSocket) 요청자(requesters)와 응답자(responders)를 지원한다. R소켓 프로토콜 개요를 포함한 자세한 내용은 스프링 프레임워크 레퍼런스의 R소켓 장을 참고하자.

10.4.1. R소켓 전략 자동 구성(RSocket Strategies Auto-configuration)

스프링 부트는 R소켓 페이로드 인코딩 및 디코딩에 필요한 모든 인프라를 제공하는 R소켓스트래티지(RSocketStrategies) 빈을 자동 구성한다. 기본적으로 자동 구성은 다음을 순서대로 구성한다.

1. 잭슨을 사용한 CBOR 코덱(CBOR codecs with Jackson)
2. 잭슨을 사용한 JSON 코덱(JSON codecs with Jackson)

spring-boot-starter-rsocket 스타터는 두 가지 의존성을 모두 제공한다. 커스텀 가능성에 대해 자세히 알아보려면 잭슨 지원 장을 참고하자.

개발자는 R소켓스트래티지커스터마이저(RSocketStrategiesCustomizer) 인터페이스를 구현하는 빈을 생성하여 R소켓스트래티지(RSocketStrategies) 컴포넌트를 커스텀할 수 있다. @Order는 코덱의 순서를 결정하므로 중요하다.

10.4.2. R소켓 서버 자동 구성(RSocket server Auto-configuration)

스프링 부트는 R소켓(RSocket) 서버 자동 구성을 제공한다. 필요한 의존성은 spring-boot-starter-rsocket에 의해 제공된다.

스프링 부트를 사용하면 웹플럭스 서버에서 웹소켓(WebSocket)을 통해 R소켓(RSocket)을 노출하거나 독립적인 R소켓 서버를 설정할 수 있다. 이는 애플리케이션 타입과 해당 구성에 따라 다르다.

웹플럭스 애플리케이션(WebApplicationType.REACTIVE 타입)의 경우 R소켓 서버는 다음 프로퍼티스가 일치하는 경우에만 웹 서버에 연결된다.

프로퍼티스(Properties)

spring.rsocket.server.mapping-path=/rsocket
spring.rsocket.server.transport=websocket

Yaml

spring:
  rsocket:
    server:
      mapping-path: "/rsocket"
      transport: "websocket"

R소켓(RSocket)을 웹 서버에 연결하는 것은 R소켓 자체가 해당 라이브러리로 구축되었기 때문에 리액터 네티(Netty)에서만 지원된다.

또는, R소켓 TCP 또는 웹소켓 서버가 독립적인 임베디드 서버로 시작된다. 의존성 요구 사항 외에도 유일한 필수 구성은 해당 서버에 대한 포트를 정의하는 것이다.

프로퍼티스(Properties)

spring.rsocket.server.port=9898

Yaml

spring:
  rsocket:
    server:
      port: 9898

10.4.3. 스프링 메세징 R소켓 지원(Spring Messaging RSocket support)

스프링 부트는 R소켓용 스프링 메시징 인프라를 자동 구성한다.

이는 스프링 부트가 애플리케이션에 대한 R소켓 요청을 처리할 R소켓메시지핸들러(RSocketMessageHandler) 빈을 생성한다는 것을 의미한다.

10.4.4. R소켓리퀘스터를 사용하여 R소켓 서비스 호출(Calling RSocket Services with RSocketRequester)

서버와 클라이언트 사이에 R소켓 채널이 설정되면 모든 당사자는 상대방에게 요청을 보내거나 받을 수 있다.

서버로서 R소켓 @Controller의 핸들러 메소드에 R소켓리퀘스터(RSocketRequester) 인스턴스를 주입할 수 있다. 클라이언트로서 먼저 R소켓 커넥션을 구성하고 설정해야 한다. 스트링 부트는 이러한 경우에 예상되는 코덱을 사용하여 RSocketRequester.Builder를 자동 구성하고 R소켓커넥터컨피규어러(RSocketConnectorConfigurer) 빈을 적용한다.

RSocketRequester.Builder 인스턴스는 프로토타입 빈이다. 즉, 각 주입 지점이 새 인스턴스를 제공한다는 의미다. 이 빌더는 상태 저장형이므로 동일한 인스턴스를 사용하여 다른 설정으로 요청자를 생성하면 안 되므로 이는 의도적으로 수행된다.

다음 코드는 일반적인 예제를 보여준다.

자바

import reactor.core.publisher.Mono;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.stereotype.Service;

@Service
public class MyService {
    private final RSocketRequester rsocketRequester;
    
    public MyService(RSocketRequester.Builder rsocketRequesterBuilder) {
        this.rsocketRequester = rsocketRequesterBuilder.tcp("example.org", 9898);
    }

    public Mono<User> someRSocketCall(String name) {
        return this.rsocketRequester.route("user").data(name).retrieveMono(User.class);
    }
}

코틀린

import org.springframework.messaging.rsocket.RSocketRequester
import org.springframework.stereotype.Service
import reactor.core.publisher.Mono

@Service
class MyService(rsocketRequesterBuilder: RSocketRequester.Builder) {
    private val rsocketRequester: RSocketRequester
    
    init {
        rsocketRequester = rsocketRequesterBuilder.tcp("example.org", 9898)
    }

    fun someRSocketCall(name: String): Mono<User> {
        return rsocketRequester.route("user").data(name).retrieveMono(
            User::class.java
        )
    } 
}

10.5. 스프링 인테그레이션(Spring Integration)

스프링 부트는 spring-boot-starter-integration “스타터(Starter)”를 포함하여 스프링 인테그레이션 작업에 대한 여러 가지 편의를 제공한다. 스프링 인테그레이션은 메시징과 HTTP, TCP 등과 같은 기타 전송에 대한 추상화를 제공한다. 클래스패스에서 스프링 인테그레이션을 사용할 수 있는 경우 @EnableIntegration 어노테이션을 통해 초기화된다.

스프링 인테그레이션 폴링 로직은 자동 구성된 태스크스케줄러(TaskScheduler)에 의존한다. 기본 폴러메타데이터(PollerMetadata)(매초 무제한의 메시지 폴링)는 spring.integration.poller.* 구성 프로퍼티스를 사용하여 커스텀할 수 있다.

스프링 부트는 또한 추가 스프링 인테그레이션 모듈이 있을 때 트리거되는 일부 기능을 구성한다. spring-integration-jmx도 클래스패스에 있으면 메시지 처리 통계가 JMX를 통해 게시된다. spring-integration-jdbc를 사용할 수 있는 경우 다음 줄에 표시된 대로 시작 시 기본 데이터베이스 스키마를 생성할 수 있다.

프로퍼티스(Properties)

spring.integration.jdbc.initialize-schema=always

Yaml

spring:
  integration:
    jdbc:
      initialize-schema: "always"

spring-integration-rsocket을 사용할 수 있는 경우 개발자는 “spring.rsocket.server.*” 프로퍼티스를 사용하여 R소켓 서버를 구성하고 인테그레이션R소켓엔드포인트(IntegrationRSocketEndpoint) 또는 R소켓아웃바운드게이트웨이(RSocketOutboundGateway) 컴포넌트를 사용하여 수신 R소켓 메시지를 처리하도록 할 수 있다. 이 인프라는 스프링 인테그레이션 R소켓 채널 어댑터 및 @MessageMapping 핸들러(“spring.integration.rsocket.server.message-mapping-enabled”가 구성된 경우)를 처리할 수 있다.

스프링 부트는 구성 프로퍼티스를 사용하여 클라이언드R소켓커넥터(ClientRSocketConnector)를 자동 구성할 수도 있다.

프로퍼티스(Properties)

# Connecting to a RSocket server over TCP
spring.integration.rsocket.client.host=example.org
spring.integration.rsocket.client.port=9898

Yaml

# Connecting to a RSocket server over TCP
spring:
  integration:
    rsocket:
      client:
        host: "example.org"
        port: 9898

프로퍼티스(Properties)

# 웹소켓을 통해 R소켓 서버에 연결
spring.integration.rsocket.client.uri=ws://example.org

Yaml

# 웹소켓을 통해 R소켓 서버에 연결
spring:
  integration:
    rsocket:
      client:
        uri: "ws://example.org"

자세한 내용은 인테그레이션오토컨피규레이션(IntegrationAutoConfiguration) 및 인테그레이션프로퍼티스(IntegrationProperties) 클래스를 참고하자.

10.6. 웹소켓(WebSockets)

스프링 부트는 임베디드 톰켓, 제티 및 언더토우에 대한 웹소켓 자동 구성을 제공한다. 독립형 컨테이너에 war 파일을 배포하면 스프링 부트는 컨테이너가 웹소켓 지원 구성을 담당한다고 가정한다.

스프링 프레임워크는 spring-boot-starter-websocket 모듈을 통해 쉽게 접근할 수 있는 MVC 웹 애플리케이션에 대한 풍부한 웹소켓 지원을 제공한다.

웹소켓 지원은 리액티브 웹 애플리케이션에서도 사용할 수 있으며 spring-boot-starter-webflux와 함께 웹소켓 API를 포함해야 한다.

<dependency>
    <groupId>jakarta.websocket</groupId>
    <artifactId>jakarta.websocket-api</artifactId>
</dependency>

10.7. 다음에 읽을 내용(What to Read Next)

다음 장에서는 애플리케이션에서 IO 기능을 활성화하는 방법을 설명한다. 이 장에서는 캐싱, 메일, 유효성 검사, 레스트(rest) 클라이언트 등에 대해 읽을 수 있다.