Real-Time Messaging with RabbitMQ and Spring Boot

Inspired by a sci‑fi series, the author explores whether modern backend tools can provide real‑time communication similar to the fictional space‑station. RabbitMQ and Spring Boot are chosen to implement a messaging system for the "Tycho" space‑station, enabling ships to send periodic updates and engage in one‑to‑one and broadcast communication.

Problem Scenario

The Tycho space‑station must monitor parameters from various spacecraft and send commands. Each ship periodically sends update messages and can also communicate directly with the station.

Use Cases

Ships send periodic update messages to the station.

One‑to‑one real‑time messaging between each ship and the station.

The station broadcasts a public message to all ships.

These use cases are realized by configuring different RabbitMQ exchanges. Because communication is bidirectional, every ship and the station act as both producers and consumers.

Exchanges route messages to queues based on routing keys; different exchange types determine how routing keys are used.

Configuration files (application.yml) define ship and station properties, including exchange names and routing keys.

## 飞船程序的 application.yml 文件
## 每艘飞船的属性值都需要更改

ship:
  name: rocinante
  update-freq: 1000

broker:
  exchange:
    direct:
      ship:
        name: rocinante-direct-exchange
        routing-key: __rocinante
      station:
        name: tyco-direct-exchange
        routing-key: __scheduled-update
    fanout:
      name: tyco-fanout-exchange
    queue:
      name: rocinante

## 空间站程序的 application.yml 文件

station:
  name: Tyco

broker:
  exchange:
    direct:
      name: tyco-direct-exchange
      routing-key: __scheduled-update
      queue:
        auto-queue: auto-queue
    fanout:
      name: tyco-fanout-exchange

Periodic Updates to the Station

A direct exchange with exact routing‑key matching is used. Ships send updates using @EnableScheduling and @Scheduled annotations.

Parameters{x=0.9688891, y=0.82120174, z=0.6792371, fuelPercentage=0.2711178}

Scheduler implementation:

@Component
@EnableScheduling
public class UpdateScheduler {
    @Value("${ship.name}")
    private String shipName;

    @Value("${broker.exchange.direct.station.name}")
    private String directExchange;

    @Value("${broker.exchange.direct.station.routing-key}")
    private String directExchangeRoutingKey;

    private Long shipUpdateFrequency;

    @Value("${ship.update-freq}")
    private void setShipUpdateFrequency(String frequency) {
        this.shipUpdateFrequency = Long.parseLong(frequency);
    }

    @Autowired
    private final RabbitTemplate rabbitTemplate;

    @SneakyThrows
    @Scheduled(fixedDelay = 1)
    public void sendUpdates() {
        String updateMessage = shipName + ": Update at " + new Date() + " " + ParameterFactory.getParameter();
        rabbitTemplate.convertAndSend(directExchange, directExchangeRoutingKey, updateMessage);
        Thread.sleep(shipUpdateFrequency);
    }
}

The station receives messages by configuring a direct exchange, binding a queue, and defining a callback method.

@Configuration
public class BrokerConfiguration {
    static String directExchangeQueue;
    static String directExchange;
    static String directRoutingKey;

    @Value("${broker.exchange.direct.routing-key}")
    private void setDirectRoutingKey(String routingKey) { BrokerConfiguration.directRoutingKey = routingKey; }

    @Value("${broker.exchange.direct.name}")
    private void setDirectExchange(String exchangeName) { BrokerConfiguration.directExchange = exchangeName; }

    @Value("${broker.exchange.direct.queue.auto-queue}")
    private void setQueueName(String queueName) { BrokerConfiguration.directExchangeQueue = queueName; }

    @Bean
    DirectExchange directExchange() { return new DirectExchange(BrokerConfiguration.directExchange); }

    @Bean
    Queue directExchangeQueue() { return new Queue(BrokerConfiguration.directExchangeQueue); }

    @Bean
    Binding updateQueueBinding(Queue directExchangeQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(directExchangeQueue).to(directExchange).with(BrokerConfiguration.directRoutingKey);
    }
}

@Configuration
public class MessageListenerConfiguration {
    @Autowired
    private final BrokerConfiguration brokerConfiguration;

    @Bean
    MessageListenerAdapter listenerAdapter(MessageHandler messageHandler) {
        return new MessageListenerAdapter(messageHandler, "receiveMessage");
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(brokerConfiguration.directExchangeQueue);
        container.setMessageListener(listenerAdapter);
        return container;
    }
}

@Component
public class MessageHandler {
    public void receiveMessage(String message) {
        System.out.println("> " + message);
    }
}

Console output example:

> rocinante: Update at Sat Jul 31 17:35:15 CDT 2021 Parameters{x=0.9688891, y=0.82120174, z=0.6792371, fuelPercentage=0.2711178}

One‑to‑One Communication Between Ship and Station

The station sends direct messages to individual ships using distinct routing keys. Example message format:

@rocinante: Go to Mars
@razorback: Go to Ceres
@nauvoo: Go to Earth

Message sending logic:

@Configuration
public class ChatInterface implements CommandLineRunner {
    private Scanner scanner;
    private final MessageHandler messageHandler;

    public ChatInterface(MessageHandler messageHandler) {
        this.messageHandler = messageHandler;
        this.scanner = new Scanner(System.in);
    }

    @Override
    public void run(String... args) {
        System.out.println("Send message...");
        while (true) {
            String msg = scanner.nextLine();
            if (msg.contains(":")) {
                messageHandler.sendMessage(msg);
            } else {
                System.out.println("Message format not correct!!");
            }
        }
    }
}

@Component
public class MessageHandler {
    @Autowired
    private final RabbitTemplate rabbitTemplate;

    public void sendMessage(String cmd) {
        String to = cmd.split(":")[0];
        String msg = cmd.split(":")[1];
        switch (to) {
            case "@rocinante":
                rabbitTemplate.convertAndSend("rocinante-direct-exchange", "__rocinante", "Station-021: " + msg);
                break;
            case "@razorback":
                rabbitTemplate.convertAndSend("razorback-direct-exchange", "__razorback", "Station-O21: " + msg);
                break;
            case "@nauvoo":
                rabbitTemplate.convertAndSend("nauvoo-direct-exchange", "__nauvoo", "Station-O21: " + msg);
                break;
            default:
                System.out.println("Message format not correct!!");
        }
    }
}

Ships receive messages via their own direct exchange and queue configuration, similar to the station setup.

Broadcasting to All Ships

A fanout exchange is used to ignore routing keys and deliver a message to every bound queue. Broadcast message format:

@all: Come back to station

Broadcast handling added to MessageHandler :

@Component
public class MessageHandler {
    @Autowired
    private final RabbitTemplate rabbitTemplate;

    public void sendMessage(String cmd) {
        String to = cmd.split(":")[0];
        String msg = cmd.split(":")[1];
        switch (to) {
            // ... other cases ...
            case "@all":
                rabbitTemplate.convertAndSend("tyco-fanout-exchange", "", "Station: " + msg);
                break;
            default:
                System.out.println("Message format not correct!!");
        }
    }
}

Ships bind a common queue to the fanout exchange:

@Configuration
public class FanoutExchangeConfiguration {
    private static String fanoutExchange;

    @Value("${broker.exchange.fanout.name}")
    private void setFanoutExchange(String fanoutExchange) { FanoutExchangeConfiguration.fanoutExchange = fanoutExchange; }

    @Bean
    FanoutExchange fanoutExchange() { return new FanoutExchange(FanoutExchangeConfiguration.fanoutExchange); }
}

@Configuration
public class BrokerConfiguration {
    // ... other beans ...
    @Bean
    Binding bindingToFanoutExchange(Queue commonQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(commonQueue).to(fanoutExchange);
    }
}

Conclusion

Each ship and the station act as both producers and consumers, requiring their own queues. The station uses a single direct exchange and queue for real‑time and periodic messages, while ships need both a direct and a fanout exchange but share one queue bound to both. The full implementation is available on GitHub.