如何使用Spring Integration DSL向ActiveMQ定时发布消息

前置依赖确认 #

确保你的项目已引入以下依赖（以Maven为例）：

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

核心实现代码 #

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.jms.dsl.Jms;
import jakarta.jms.ConnectionFactory;
import java.time.LocalDateTime;

@Configuration
public class TimedMessageFlowConfig {

    @Bean
    public IntegrationFlow scheduledSendToMyQueue(ConnectionFactory activeMqConnectionFactory) {
        return IntegrationFlow
                // 每隔10秒生成一条消息
                .fromSupplier(
                        // 自定义消息内容，可根据业务需求修改
                        () -> "定时消息生成时间：" + LocalDateTime.now(),
                        e -> e.poller(Pollers.fixedDelay(10000))
                )
                // 发送到my.queue队列
                .handle(Jms.outboundAdapter(activeMqConnectionFactory)
                        .destination("my.queue")
                )
                .get();
    }
}

参数说明 #

• 定时规则：Pollers.fixedDelay(10000) 单位为毫秒，10000对应10秒间隔，若需要固定速率触发可替换为Pollers.fixedRate(10000)
• 消息内容：fromSupplier的第一个参数为消息生产逻辑，可自定义为任意类型的业务数据，若需要自定义消息头可使用MessageBuilder构造消息对象
• 队列配置：destination("my.queue")直接指定目标队列名即可，Spring会自动对接ActiveMQ的队列资源

可选配置（application.yml） #

若你的ActiveMQ不是本地默认配置，可在配置文件中修改连接参数：

spring:
  activemq:
    broker-url: tcp://你的ActiveMQ服务地址:61616
    user: 连接用户名
    password: 连接密码

注意：如果需要发送自定义对象类型的消息，需要保证对象实现Serializable接口，也可根据需求配置全局JMS消息转换器

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