RSocket in Spring Boot: Reactive Socket Communication for Real-Time Apps

You've built REST APIs, maybe even dabbled with WebSockets for real-time chat. But when your Spring Boot application needs to handle thousands of concurrent streaming connections with backpressure — think a real-time analytics dashboard consuming market data from multiple sources — HTTP starts to show its age. The overhead of headers, the lack of native streaming, the pain of polling. Enter RSocket, a protocol that was designed from the ground up for reactive, multiplexed, bidirectional communication over a single TCP connection. Spring Boot 3.2+ (and even 2.7 with some tuning) provides first-class support for RSocket via spring-boot-starter-rsocket. In this article, I'll walk you through setting up an RSocket server and client for a SaaS billing system that streams usage metrics and handles payment events. We'll cover all four interaction models, dive into backpressure with Reactor Flux, and I'll share a production incident where a missing keepalive caused a silent connection drop in our analytics pipeline. By the end, you'll have a working RSocket setup and know what the official docs gloss over — like thread pool sizing, error handling with metadata, and debugging with WireShark.

Setting Up RSocket Server and Client in Spring Boot

Let's start by adding the RSocket starter to your Spring Boot project. If you're using Maven, add this to your pom.xml:

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-rsocket</artifactId> </dependency>

For Gradle: implementation 'org.springframework.boot:spring-boot-starter-rsocket'

This brings in spring-boot-starter-webflux (since RSocket is reactive) and the RSocket Java library. Now configure the server in application.properties. The simplest setup is a TCP server:

spring.rsocket.server.port=7000 spring.rsocket.server.transport=tcp

For a WebSocket server (useful if you need to go through a load balancer that doesn't speak TCP):

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

Now create a controller to handle RSocket routes. Unlike HTTP controllers, you don't use @RequestMapping; you use @MessageMapping and @ConnectMapping. Here's a simple server that echoes messages:

@Controller public class RSocketController {

@MessageMapping("echo") public Mono<String> echo(String message) { return Mono.just("Echo: " + message); } }

The client side is a bit more involved. You need an RSocketRequester bean that connects to the server. Here's a configuration class:

@Configuration public class RSocketClientConfig {

@Bean RSocketRequester rSocketRequester(RSocketRequester.Builder builder) { return builder .connectTcp("localhost", 7000) .block(); } }

Then inject RSocketRequester into your service and call routes:

@Service public class BillingService { private final RSocketRequester requester;

public BillingService(RSocketRequester requester) { this.requester = requester; }

public Mono<String> callEcho(String message) { return requester.route("echo").data(message).retrieveMono(String.class); } }

That's the basic setup. But in production, you'll want to configure keepalive, retry, and load balancing. I'll cover those in subsequent sections.

@Controller
public class RSocketController {

    @MessageMapping("echo")
    public Mono<String> echo(String message) {
        return Mono.just("Echo: " + message);
    }

    @MessageMapping("fire-and-forget")
    public Mono<Void> fireAndForget(String event) {
        System.out.println("Received event: " + event);
        return Mono.empty();
    }

    @MessageMapping("stream")
    public Flux<String> stream(String topic) {
        return Flux.interval(Duration.ofSeconds(1))
            .map(i -> topic + " update #" + i);
    }
}

What the Official Docs Won't Tell You

The Spring Boot reference guide tells you how to set up RSocket, but it glosses over three critical things I've learned the hard way. First, thread pool sizing. RSocket uses Reactor's event loop by default. If you have a blocking operation in a @MessageMapping handler (like a JDBC call), you'll block the event loop and kill throughput. Always offload blocking work to a bounded elastic scheduler using .subscribeOn(Schedulers.boundedElastic()). Second, error handling with metadata. The RSocket spec allows sending metadata alongside data, but Spring Boot's @MessageMapping by default only extracts data. If your client sends error codes in metadata, you need a custom MetadataExtractor. Third, connection multiplexing. RSocket allows multiple streams over one connection, but if your server uses a single connection per client, a slow stream can block others. Use RSocket's load balancer client (spring.rsocket.client.load-balancer) to pool connections. I once saw a production incident where a single slow query on one route stalled all other requests because they shared the same TCP connection. We fixed it by using a connection pool with min idle connections. The official docs don't mention any of this. Also, be aware that RSocket over WebSocket requires a different path for load balancer health checks. Most load balancers (like Nginx) don't understand RSocket's binary protocol, so you need a separate HTTP health endpoint.

@MessageMapping("blocking-query")
public Mono<String> handleBlockingQuery(String query) {
    return Mono.fromCallable(() -> {
        // Simulate JDBC call
        Thread.sleep(1000);
        return "Result for: " + query;
    }).subscribeOn(Schedulers.boundedElastic());
}

The Four Interaction Models: Fire-and-Forget, Request-Response, Request-Stream, Channel

RSocket's killer feature is its four interaction models, all over a single connection. Let's implement each one in our billing system. Fire-and-forget is for events where you don't need a response, like logging a payment attempt. Request-response is for queries like 'get invoice details'. Request-stream is for subscribing to a live feed, like real-time usage metrics. Channel is for bidirectional streaming, like a chat between two microservices. Here's the server-side implementation:

@Controller public class BillingController {

// Fire-and-forget: log payment event @MessageMapping("payment.event") public Mono<Void> logPaymentEvent(PaymentEvent event) { return Mono.fromRunnable(() -> paymentLogger.log(event)); }

// Request-response: get invoice @MessageMapping("invoice.get") public Mono<Invoice> getInvoice(@Payload String invoiceId) { return invoiceService.findById(invoiceId); }

// Request-stream: subscribe to usage metrics for a tenant @MessageMapping("usage.stream") public Flux<UsageMetric> streamUsage(@Payload String tenantId) { return usageService.streamMetrics(tenantId); }

// Channel: bidirectional chat for support @MessageMapping("support.chat") public Flux<ChatMessage> chat(Flux<ChatMessage> clientMessages) { return clientMessages.map(msg -> new ChatMessage("Bot", "You said: " + msg.text())); } }

On the client side, you use different methods on RSocketRequester:

// Fire-and-forget requester.route("payment.event").data(event).send().subscribe();

// Request-response Mono<Invoice> invoice = requester.route("invoice.get").data("INV-123").retrieveMono(Invoice.class);

// Request-stream Flux<UsageMetric> metrics = requester.route("usage.stream").data("tenant-1").retrieveFlux(UsageMetric.class);

// Channel Flux<ChatMessage> replies = requester.route("support.chat").data(clientMessageFlux).retrieveFlux(ChatMessage.class);

Notice that for channel, you pass a Flux as data. This is where RSocket shines over WebSocket: you get backpressure out of the box. If the server is slow, the client's Flux will be requested with backpressure signals.

@Controller
public class BillingController {

    @MessageMapping("payment.event")
    public Mono<Void> logPaymentEvent(@Payload PaymentEvent event) {
        return Mono.fromRunnable(() -> System.out.println("Logging: " + event));
    }

    @MessageMapping("invoice.get")
    public Mono<Invoice> getInvoice(@Payload String invoiceId) {
        return Mono.just(new Invoice(invoiceId, 100.0));
    }

    @MessageMapping("usage.stream")
    public Flux<UsageMetric> streamUsage(@Payload String tenantId) {
        return Flux.interval(Duration.ofSeconds(1))
            .map(i -> new UsageMetric(tenantId, i, Math.random() * 100));
    }

    @MessageMapping("support.chat")
    public Flux<ChatMessage> chat(Flux<ChatMessage> clientMessages) {
        return clientMessages.map(msg -> new ChatMessage("Bot", "Echo: " + msg.text()));
    }
}

Backpressure and Flow Control in RSocket

Backpressure is the ability of a consumer to signal to the producer how much data it can handle. In RSocket, this is built into the protocol itself, unlike WebSocket where you'd have to implement your own flow control. Spring Boot's RSocket uses Reactor's Flux and Mono, which support backpressure natively. When a client calls .retrieveFlux(), it sends a REQUEST_N frame to the server with an initial demand (default 256). The server then sends that many items. As the client processes items, it sends more REQUEST_N frames. This prevents the server from overwhelming the client. In our billing system, imagine a client that streams usage metrics but has a slow database write. The client's Flux subscriber can request one item at a time, and the server will wait. Here's how to control demand on the client side:

Flux<UsageMetric> metrics = requester.route("usage.stream") .data("tenant-1") .retrieveFlux(UsageMetric.class) .limitRate(10); // Request 10 items at a time

On the server side, you can also control backpressure by using Flux.create() with a custom overflow strategy:

@MessageMapping("usage.stream") public Flux<UsageMetric> streamUsage(@Payload String tenantId) { return Flux.create(sink -> { // Simulate external data source ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); executor.scheduleAtFixedRate(() -> { sink.next(new UsageMetric(tenantId, System.currentTimeMillis(), Math.random() * 100)); }, 0, 1, TimeUnit.SECONDS); sink.onCancel(() -> executor.shutdown()); }, FluxSink.OverflowStrategy.LATEST); // Drop old data if client is slow }

OverflowStrategy.LATEST ensures that if the client is slow, we only keep the latest metric, not buffer indefinitely. This is critical for real-time analytics where stale data is worse than missing data.

Flux<UsageMetric> metrics = requester
    .route("usage.stream")
    .data("tenant-1")
    .retrieveFlux(UsageMetric.class)
    .limitRate(10) // Request 10 at a time
    .doOnNext(metric -> {
        // Simulate slow processing
        Thread.sleep(100);
        System.out.println("Processed: " + metric);
    })
    .subscribe();

Error Handling and Resilience with RSocket

Errors in RSocket can happen at different levels: connection failure, route not found, or application exception. Spring Boot's RSocket support maps exceptions to error frames with metadata. By default, if your @MessageMapping throws an exception, the server sends an error frame to the client with the exception message. But in production, you want more control. First, use @MessageExceptionHandler to handle exceptions in a centralized way:

@ControllerAdvice public class RSocketExceptionHandler {

@MessageExceptionHandler public Mono<String> handle(IllegalArgumentException e) { return Mono.just("Error: " + e.getMessage()); }

@MessageExceptionHandler public Mono<Void> handleGeneric(Exception e) { return Mono.error(new RuntimeException("Internal error")); } }

Second, on the client side, you need to handle errors gracefully. RSocketRequester methods return Mono or Flux, so you can use Reactor's error operators:

Mono<Invoice> invoice = requester.route("invoice.get") .data("INV-123") .retrieveMono(Invoice.class) .onErrorResume(e -> { log.error("Failed to get invoice", e); return Mono.just(new Invoice("UNKNOWN", 0.0)); }) .timeout(Duration.ofSeconds(5));

Third, connection resilience. If the RSocket connection drops, you need to reconnect. Spring Boot's RSocketRequester.Builder has a .reconnect() method:

@Bean RSocketRequester rSocketRequester(RSocketRequester.Builder builder) { return builder .rsocketConnector(connector -> connector .reconnect(Retry.fixedDelay(10, Duration.ofSeconds(5))) .keepalive(Duration.ofSeconds(20), Duration.ofSeconds(90))) .connectTcp("localhost", 7000) .block(); }

This will retry connection 10 times with a 5-second delay. But be careful: if the server is down for maintenance, you'll flood logs. Use an exponential backoff instead:

Retry.backoff(5, Duration.ofSeconds(1)).maxBackoff(Duration.ofSeconds(30))

@Bean
RSocketRequester rSocketRequester(RSocketRequester.Builder builder) {
    return builder
        .rsocketConnector(connector -> connector
            .reconnect(Retry.backoff(5, Duration.ofSeconds(1))
                .maxBackoff(Duration.ofSeconds(30)))
            .keepalive(Duration.ofSeconds(20), Duration.ofSeconds(90)))
        .connectTcp("localhost", 7000)
        .block();
}

Security: Authentication and Authorization in RSocket

RSocket doesn't have built-in authentication like HTTP Basic Auth or JWT. You need to implement it yourself using metadata. The common pattern is to send an auth token in the metadata of the first request (or during connection setup via @ConnectMapping). On the server side, you create a MetadataExtractor that extracts the token and validates it. Here's how to set up JWT authentication for RSocket. First, configure the client to send a token in metadata:

@Bean RSocketRequester rSocketRequester(RSocketRequester.Builder builder, JwtTokenProvider tokenProvider) { return builder .rsocketConnector(connector -> connector .metadataMimeType(MimeTypeUtils.APPLICATION_JSON) .metadataExtractor(new MyMetadataExtractor())) .setupMetadata(tokenProvider.generateToken(), MimeTypeUtils.APPLICATION_JSON) .connectTcp("localhost", 7000) .block(); }

On the server, create a @ConnectMapping handler that validates the setup token:

@Controller public class SecurityController {

@ConnectMapping public Mono<Void> onConnect(RSocketRequester requester, @Payload String token) { if (!jwtValidator.isValid(token)) { return Mono.error(new SecurityException("Invalid token")); } // Store authenticated requester in a registry return Mono.empty(); } }

If the token is invalid, the server rejects the connection by throwing an exception. The client will receive an error and can retry with a new token. For per-request authorization, you can extract the token from metadata in each @MessageMapping handler. But that's tedious. Instead, use a WebFilter-like approach with RSocket's interceptor. Spring Boot doesn't have a built-in RSocket interceptor, but you can use the RSocketStrategies to add a custom SocketAcceptorInterceptor. Here's a simplified version:

@Bean RSocketStrategies rSocketStrategies() { return RSocketStrategies.builder() .socketAcceptorInterceptor(new AuthInterceptor()) .build(); }

In production, you'd integrate with Spring Security's reactive authentication manager.

@Controller
public class SecurityController {

    @ConnectMapping
    public Mono<Void> onConnect(RSocketRequester requester, @Payload String token) {
        if (!"valid-token".equals(token)) {
            return Mono.error(new SecurityException("Invalid token"));
        }
        System.out.println("Client connected with valid token");
        return Mono.empty();
    }

    @MessageMapping("secure.endpoint")
    public Mono<String> secureEndpoint(@Payload String data) {
        return Mono.just("Secure data: " + data);
    }
}

Testing RSocket Endpoints with Spring Boot Test

Testing RSocket endpoints requires a different approach than REST controllers. You can't use MockMvc; you need a real RSocket client connected to a test server. Spring Boot provides @SpringBootTest with a random port for RSocket. Here's how to write a test for our billing controller:

@SpringBootTest(properties = { "spring.rsocket.server.port=0" // random port }) class BillingControllerTest {

@Autowired private RSocketRequester.Builder builder;

private RSocketRequester requester;

@BeforeEach void setUp() { requester = builder .connectTcp("localhost", 0) // Will be replaced by actual port .block(); }

@Test void testRequestResponse() { Mono<Invoice> result = requester.route("invoice.get") .data("INV-123") .retrieveMono(Invoice.class); StepVerifier.create(result) .expectNextMatches(invoice -> invoice.id().equals("INV-123")) .verifyComplete(); }

@Test void testRequestStream() { Flux<UsageMetric> stream = requester.route("usage.stream") .data("tenant-1") .retrieveFlux(UsageMetric.class) .take(3); StepVerifier.create(stream) .expectNextCount(3) .verifyComplete(); }

@Test void testFireAndForget() { Mono<Void> result = requester.route("payment.event") .data(new PaymentEvent("user-1", 100.0)) .send(); StepVerifier.create(result) .verifyComplete(); } }

Note: The random port assignment with spring.rsocket.server.port=0 works, but you need to inject the actual port. Use @LocalServerPort for HTTP, but for RSocket, you can use a test property or a custom listener. A simpler approach is to use a fixed port in test profile. Also, use @AutoConfigureRSocket to auto-configure the RSocket client for testing. For integration tests with Testcontainers, you can start an RSocket server in a container, but that's rare. Most RSocket tests are integration tests with the server embedded.

@SpringBootTest(properties = {"spring.rsocket.server.port=0"})
@AutoConfigureRSocket
class BillingControllerTest {

    @Autowired
    private RSocketRequester.Builder builder;

    private RSocketRequester requester;

    @BeforeEach
    void setUp() {
        requester = builder
            .connectTcp("localhost", 0)
            .block();
    }

    @Test
    void testInvoiceRequestResponse() {
        StepVerifier.create(
            requester.route("invoice.get")
                .data("INV-123")
                .retrieveMono(Invoice.class)
        ).expectNextMatches(i -> i.id().equals("INV-123"))
         .verifyComplete();
    }
}

Production Monitoring and Debugging RSocket Connections

In production, you need to monitor RSocket connections and debug issues. Spring Boot Actuator doesn't expose RSocket-specific metrics by default, but you can add them with Micrometer. First, add the Micrometer RSocket module (not part of Spring Boot starter):

<dependency> <groupId>io.micrometer</groupId> <artifactId>micrometer-core</artifactId> </dependency> <dependency> <groupId>io.rsocket</groupId> <artifactId>rsocket-micrometer</artifactId> <version>1.1.4</version> </dependency>

Then configure the RSocket connector to use Micrometer:

@Bean RSocketRequester rSocketRequester(RSocketRequester.Builder builder, MeterRegistry meterRegistry) { return builder .rsocketConnector(connector -> connector .interceptors(interceptors -> interceptors .add(new MicrometerRSocketInterceptor(meterRegistry))) .keepalive(Duration.ofSeconds(20), Duration.ofSeconds(90))) .connectTcp("localhost", 7000) .block(); }

Now you'll get metrics like rsocket.connections, rsocket.streams.active, and rsocket.bytes.sent. Export them to Prometheus via micrometer-registry-prometheus. For debugging, use WireShark to capture RSocket frames. RSocket uses a binary protocol, but WireShark can decode it with the RSocket dissector (install via Lua script). Common issues to look for: RESUME frames (connection resumption), KEEPALIVE frames (are they being sent?), and ERROR frames. Also, enable RSocket logging in Spring Boot:

logging.level.io.rsocket=DEBUG

This will log frame-level details. But be careful: in production, this can generate huge logs. Use it temporarily during incidents. Another tip: if you're using RSocket over WebSocket, you can debug with browser dev tools (Network tab) since WebSocket frames are visible. For TCP, you need WireShark. I once spent two hours debugging a mysterious latency spike only to find that a firewall was delaying KEEPALIVE frames. WireShark showed the delay immediately.

@Configuration
public class MonitoringConfig {

    @Bean
    RSocketRequester rSocketRequester(RSocketRequester.Builder builder, MeterRegistry meterRegistry) {
        return builder
            .rsocketConnector(connector -> connector
                .interceptors(interceptors -> interceptors
                    .add(new MicrometerRSocketInterceptor(meterRegistry)))
                .keepalive(Duration.ofSeconds(20), Duration.ofSeconds(90)))
            .connectTcp("localhost", 7000)
            .block();
    }
}