SpringBoot整合ActiveMQ

Posted 2022-11-23 libbo

目录结构

引入 maven依赖

  <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.4.RELEASE</version>
        <relativePath/> 
    </parent>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

引入 application.yml配置

spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616
    user: admin
    password: admin
queue: springboot-queue
server:
  port: 8080

创建QueueConfig

@Configuration
public class QueueConfig 
    @Value("$queue")
    private String queue;

    @Bean
    public Queue logQueue() 
        return new ActiveMQQueue(queue);
    

    @Bean
    public JmsTemplate jmsTemplate(ActiveMQConnectionFactory activeMQConnectionFactory, Queue queue) 
        JmsTemplate jmsTemplate = new JmsTemplate();
        jmsTemplate.setDeliveryMode(2);// 进行持久化配置 1表示非持久化，2表示持久化</span>
        jmsTemplate.setConnectionFactory(activeMQConnectionFactory);
        jmsTemplate.setDefaultDestination(queue); // 此处可不设置默认，在发送消息时也可设置队列
        jmsTemplate.setSessionAcknowledgeMode(4);// 客户端签收模式</span>
        return jmsTemplate;
    

    // 定义一个消息监听器连接工厂，这里定义的是点对点模式的监听器连接工厂
    @Bean(name = "jmsQueueListener")
    public DefaultJmsListenerContainerFactory jmsQueueListenerContainerFactory(
            ActiveMQConnectionFactory activeMQConnectionFactory) 
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(activeMQConnectionFactory);
        // 设置连接数
        factory.setConcurrency("1-10");
        // 重连间隔时间
        factory.setRecoveryInterval(1000L);
        factory.setSessionAcknowledgeMode(4);
        return factory;
    

创建生产者：

@SpringBootApplication
@Component
@EnableScheduling
public class Producer 
    
    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;
    
    @Autowired
    private Queue queue;
    
    @Scheduled(fixedDelay=3000)
    public void send() 
        String result = System.currentTimeMillis()+"---测试";
        System.out.println("result"+result);
        jmsMessagingTemplate.convertAndSend(queue,result);
    
    public static void main(String[] args) 
        SpringApplication.run(Producer.class, args);
    

创建消费者的application.yml

spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616
    user: admin
    password: admin
queue: springboot-queue
server:
  port: 8081

创建消费者：

@Component
@SpringBootApplication
public class consumer 

    private int count =0;
    
    @JmsListener(destination = "$queue")
    public void receive(TextMessage textMessage,Session session) throws JMSException 
        String text = textMessage.getText();
        
        System.out.println("消费:"+text+"第几次获取消息count："+(++count));
        
        System.out.println();
        String jmsMessageID = textMessage.getJMSMessageID();
    
    
    public static void main(String[] args) 
        SpringApplication.run(consumer.class,args);
    

结果显示：