如何为ActiveMQ未订阅队列设置过期及删除机制？

Hey there! Let's tackle your two ActiveMQ + Spring Boot WebSocket questions head-on, using your existing implementation as a starting point. Your setup with SimpMessagingTemplate and AbstractWebSocketMessageBrokerConfigurer looks solid—let's build on it to meet your requirements.

1. Setting Expiration for ActiveMQ Queues When No Subscribers Connect in a Specified Window #

First, let's clarify: your question refers to queue expiration (deleting the queue itself if no user subscribes within a timeframe), not just message expiration (discarding unconsumed messages). Your current code sets an expires header on messages, which handles message-level expiration—but let's focus on queue-level expiration as your question asks.

To make ActiveMQ auto-delete queues that have no subscribers for a set period, you have two reliable options:

Option 1: Configure ActiveMQ Directly (Recommended) #

Modify your ActiveMQ activemq.xml to add a policy entry targeting your user-specific queues (since /user/{userId}/reply maps to queues in ActiveMQ). This lets the broker handle cleanup automatically:

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}">
  <destinationPolicy>
    <policyMap>
      <policyEntries>
        <!-- Apply rules to all queues under the /user prefix -->
        <policyEntry queue=">/user/*" 
                     queueTTL="300000" <!-- Delete queue if inactive for 5 minutes (300,000 ms) -->
                     autoDelete="true" <!-- Enable automatic deletion when TTL is hit -->
                     expireMessagesPeriod="60000" <!-- Check for inactive queues every minute -->
                     />
      </policyEntries>
    </policyMap>
  </destinationPolicy>
  <!-- Keep your existing broker configs here -->
</broker>

• queueTTL: Defines how long a queue can stay inactive (no subscribers, no pending messages) before being deleted.
• autoDelete: Tells ActiveMQ to actually remove the queue once the TTL is reached.
• expireMessagesPeriod: How often the broker checks for inactive queues to clean up.

Option 2: Pass Queue Properties via Spring (Less Ideal) #

If you can't modify ActiveMQ's config file, you could programmatically create queues with TTL properties using ActiveMQ's JMS API—but since you're using STOMP relay, this requires extra work to map your STOMP destinations to JMS queues. The first option is far cleaner for your use case.

2. Cleaning Up Unsubscribed Queues to Prevent Performance Bloat #

For large user bases, unused queues can consume memory and slow down ActiveMQ. Here are two effective solutions:

Solution 1: Lean on ActiveMQ's Built-in Auto-Cleanup #

As covered in the first section, configuring queueTTL and autoDelete in activemq.xml is the simplest, low-maintenance approach. ActiveMQ will automatically remove queues that meet the inactive criteria without any custom code.

Solution 2: Build a Scheduled Cleanup Task (For Custom Control) #

If you need more flexibility (e.g., only clean up during off-peak hours, or add custom logic), create a Spring Boot scheduled task that uses ActiveMQ's JMX API to scan and delete unused queues:

1. First, enable JMX in ActiveMQ (add this to activemq.xml):

<managementContext>
  <managementContext createConnector="true" connectorPort="1099" connectorPath="/jmxrmi" />
</managementContext>

2. Create the scheduled task in your Spring Boot app:

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.management.MBeanServerConnection;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import javax.management.ObjectName;
import java.util.Set;

@Component
public class UnusedQueueCleaner {

    private static final String ACTIVEMQ_JMX_ENDPOINT = "service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi";

    // Run every hour (adjust the rate as needed)
    @Scheduled(fixedRate = 3600000)
    public void purgeUnusedQueues() throws Exception {
        // Connect to ActiveMQ's JMX server
        JMXConnector connector = JMXConnectorFactory.connect(new JMXServiceURL(ACTIVEMQ_JMX_ENDPOINT));
        MBeanServerConnection mBeanConn = connector.getMBeanServerConnection();

        // Query all queues in the broker
        ObjectName queueFilter = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName=*");
        Set<ObjectName> allQueues = mBeanConn.queryNames(queueFilter, null);

        for (ObjectName queue : allQueues) {
            // Check if the queue has no consumers and no pending messages
            int consumerCount = (Integer) mBeanConn.getAttribute(queue, "ConsumerCount");
            long messageCount = (Long) mBeanConn.getAttribute(queue, "QueueSize");

            if (consumerCount == 0 && messageCount == 0) {
                // Delete the queue
                mBeanConn.invoke(queue, "remove", new Object[]{}, new String[]{});
                System.out.println("Deleted inactive queue: " + queue.getKeyProperty("destinationName"));
            }
        }

        connector.close();
    }
}

Don't forget to add @EnableScheduling to your main Spring Boot application class to activate the scheduled task.

Solution 3: Use ActiveMQ Advisory Messages (Real-Time Cleanup) #

For immediate cleanup when the last subscriber disconnects, subscribe to ActiveMQ's advisory messages. These are internal messages the broker sends when consumers connect or disconnect. You can build a JMS listener that listens to these advisories and deletes the queue as soon as the last consumer leaves. This is great for real-time cleanup but requires more setup than the other options.

Quick Note on Your Existing Code #

• Your NotificationWebSocketService correctly sets message expiration via the expires header—this will delete individual messages if they're not consumed within 20 seconds, which complements queue-level expiration nicely.
• Your WebSocketConfig is properly configured to relay STOMP messages to ActiveMQ. Just confirm that ActiveMQ's STOMP port (61613) is open and the broker is reachable from your Spring Boot app.

内容的提问来源于stack exchange，提问作者user1122