如何从activemq中删除已使用的消息？

Question

我已经尝试了下面的代码。

@Component
public class JmsProducer {

    @Autowired
    JmsTemplate jmsTemplate;
    
    @Value("${gkz.activemq.queue}")
    String queue;
    
    public void send(Customer customer){
        jmsTemplate.convertAndSend(queue, customer);
    }
}

@Component
public class JmsConsumer {
    @Autowired
    private MessageStorage customerStorage;

    @JmsListener(destination = "${gkz.activemq.queue}",containerFactory="jsaFactory")
    public void receive(Customer customer){
        System.out.println("Recieved Message: " + customer);
        customerStorage.add(customer);
    }
}

@PostMapping(value="/api/customer")
public Customer postCustomer(@RequestBody Customer customer){
    jmsProducer.send(customer);
    return customer;
}

@GetMapping(value="/api/customers")
public List<Customer> getAll(){
    List<Customer> customers = customerStorage.getAll();
    return customers;
}

@DeleteMapping(value="/api/customers/clear")
public String clearCustomerStorage() {
    customerStorage.clear();
    return "Clear All CustomerStorage!";
}

public class MessageStorage {
    private List<Customer> customers = new ArrayList<>();

    public void add(Customer customer) {
        customers.add(customer);
    }

    public void clear() {
        customers.clear();
    }

    public List<Customer> getAll(){
        return customers;
    }
}

@Configuration
public class ConnectionFactoryConfiguration {
    
    @Value("${gkz.activemq.broker.url}")
    String brokerUrl;
    
    @Value("${gkz.activemq.borker.username}")
    String userName;
    
    @Value("${gkz.activemq.borker.password}")
    String password;
    
    @Bean
    public ConnectionFactory connectionFactory(){
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL(brokerUrl);
        connectionFactory.setUserName(userName);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }
    
    @Bean
    public MessageConverter jacksonJmsMessageConverter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT);
        converter.setTypeIdPropertyName("_type");
        return converter;
    }
    
    
    //Used for Receiving Message
    @Bean
    public JmsListenerContainerFactory<?> jsaFactory(ConnectionFactory connectionFactory,
                                                    DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setMessageConverter(jacksonJmsMessageConverter());
        configurer.configure(factory, connectionFactory);
        return factory;
    }
 
    
    //Used for Sending Messages.
    @Bean
    public JmsTemplate jmsTemplate(){
        JmsTemplate template = new JmsTemplate();
        template.setMessageConverter(jacksonJmsMessageConverter());
        template.setConnectionFactory(connectionFactory());
        return template;
    }
    
}

整个代码https://grokonez.com/java-integration/distributed-system/activemq-producer-consumer-springboot-restapis-example

Answer 1

单击MQ上的清除选项，然后重新启动MQ服务。这将清除所有未处于挂起状态的消息。