springboot整合activemq

一、activemq简介

Apache ActiveMQ是Apache Software Foundation（ASF）的一个开源消息代理项目，它是使用Java语言编写的消息中间件，基于JMS（Java Message Service）规范，提供了常见的消息传递模型（点对点模型和发布/订阅模型）来支持应用程序之间的通信。

二、springboot整合activemq基础

在使用springboot整合activemq之前，需要引入以下依赖。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

在application.properties或application.yml文件中配置activemq的基本属性。

spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.activemq.in-memory=true

其中，broker-url是activemq连接的地址，user和password是登录activemq的用户名和密码，in-memory表示是否使用内存存储消息，默认为false表示使用持久化存储。

在代码中注入JmsTemplate来发送和接收消息。以下是一个简单的例子。

@Service
public class Producer {
 
    @Autowired
    JmsTemplate jmsTemplate;
    
    public void sendMessage(String queueName, String message) {
        jmsTemplate.send(queueName, session -> session.createTextMessage(message));
    }
}

Producer中使用JmsTemplate发送消息，其中queueName为消息队列的名称，message为发送的消息内容。接收消息的代码如下。

@Service
public class Consumer {
 
    @JmsListener(destination = "myqueue")
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

Consumer中使用@JmsListener注解来监听消息队列，destination表示监听的队列名称，receiveMessage方法接收消息内容。

三、activemq高级用法

1. 事务支持

在消息处理过程中，需要保证消息的可靠性。使用activemq的事务支持可以满足此需求。

@Service
public class Producer {
 
    @Autowired
    JmsTemplate jmsTemplate;
    
    public void sendMessage(String queueName, String message) {
        jmsTemplate.execute(session -> {
            Message msg = session.createTextMessage(message);
            session.createProducer(session.createQueue(queueName)).send(msg);
            return null;
        });
    }
}

在发送消息的方法中，使用jmsTemplate.execute来开启一个事务，在事务中发送消息，并通过return null提交事务。如果提交成功，消息会被成功发送；否则，消息将被回滚。

2. 发布/订阅模型

在使用发布/订阅模型时，需要定义主题（topic）。

@Configuration
public class TopicConfig {
 
    @Bean
    public Topic topic() {
        return new ActiveMQTopic("mytopic");
    }
}

定义主题的代码如上所示。在Producer中发送消息时，使用convertAndSend方法，而不是send方法。

@Service
public class Producer {
 
    @Autowired
    JmsTemplate jmsTemplate;
    
    public void sendMessage(String topicName, String message) {
        jmsTemplate.convertAndSend(topicName, message);
    }
}

在Consumer中接收消息时，使用@JmsListener注解，并设置containerFactory为JmsListenerContainerFactory。

@Service
public class Consumer {
 
    @Autowired
    JmsListenerContainerFactory jmsListenerContainerFactory;
 
    @JmsListener(destination = "mytopic", containerFactory = "jmsListenerContainerFactory")
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

3. 消息过滤

activemq支持根据一定的条件过滤消息，从而只选择特定的消息进行消费。在定义JmsListener时，可以设置selector参数，该参数定义了选择哪些消息进行消费。

@Service
public class Consumer {
 
    @JmsListener(destination = "myqueue", selector = "color = 'blue'")
    public void receiveBlueMessage(String message) {
        System.out.println("Received message: " + message);
    }
 
    @JmsListener(destination = "myqueue", selector = "color = 'red'")
    public void receiveRedMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

以上代码中，JmsListener的selector参数分别设置为color=’blue’和color=’red’，表示只选择color属性为blue或red的消息进行消费。

4. 死信队列

activemq支持死信队列（Dead Letter Queue，DLQ），当消息处理失败时，可以将该消息发送到DLQ中。在使用DLQ时，需要定义主题和队列。

@Configuration
public class TopicConfig {
 
    @Bean
    public Topic topic() {
        return new ActiveMQTopic("mytopic.dlq");
    }
}
 
@Configuration
public class QueueConfig {
 
    @Bean
    public Queue queue() {
        return new ActiveMQQueue("myqueue");
    }
 
    @Bean
    public RedeliveryPolicy redeliveryPolicy() {
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(3);
        redeliveryPolicy.setInitialRedeliveryDelay(3000);
        redeliveryPolicy.setRedeliveryDelay(3000);
        redeliveryPolicy.setUseExponentialBackOff(false);
        return redeliveryPolicy;
    }
 
    @Bean
    public RedeliveryPolicyMap redeliveryPolicyMap() {
        RedeliveryPolicyMap redeliveryPolicyMap = new RedeliveryPolicyMap();
        redeliveryPolicyMap.setDefaultEntry(redeliveryPolicy());
        return redeliveryPolicyMap;
    }
 
    @Bean
    public JmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrency("3-10"); // 控制消费者个数
        factory.setSessionTransacted(true);
        factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
 
        // 配置死信队列
        RedeliveryPlugin redeliveryPlugin = new RedeliveryPlugin();
        redeliveryPlugin.setRedeliveryPolicyMap(redeliveryPolicyMap());
        RedeliveryPlugin[] plugins = { redeliveryPlugin };
        factory.setPlugins(plugins);
        return factory;
    }
}

以上代码中，定义了主题为mytopic.dlq的死信队列，在QueueConfig中配置了队列，并定义了RedeliveryPolicy和RedeliveryPolicyMap，用于配置消息重发的策略。在JmsListenerContainerFactory中设置了sessionTransacted为true，sessionAcknowledgeMode为CLIENT_ACKNOWLEDGE，表示使用事务和客户端确认模式，并通过RedeliveryPlugin来将未处理的消息发送到DLQ中。

四、总结

本文介绍了springboot整合activemq的基础和高级用法，包括事务支持、发布/订阅模型、消息过滤和死信队列。通过本文的学习，读者可以更加深入地了解activemq的应用场景和使用方法。