如何从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!";
}

消息存储

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 服务。这将清除所有未处于待处理状态的消息。