如何从activemq中移除消费消息？

Question

​

我需要在消费完消息后立即从activemq中删除它。例如，我向队列发送一条消息，然后使用它，并需要将其从队列中删除。我在这里使用了一个消息存储和一个clear()方法。消息被添加到队列中，并且只从消息存储中删除，而不从队列中删除。我需要一种从队列中删除消息的方法。感谢你的帮助！

我试过下面的代码。

生产者

@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!";
}

MessageStorage

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;
    }
}

ConnectionFactoryConfiguration

@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上的purge选项，稍后重新启动MQ服务。这将清除所有未处于挂起状态的消息。