SpringBoot整合ActiveMQ

标签：cto   发送   over   value   ima   class   tar   依赖   conf   

目录结构

引入 maven依赖

  org.springframework.boot
        spring-boot-starter-parent
        1.5.4.RELEASEUTF-8UTF-81.8org.springframework.boot
            spring-boot-starter
        org.springframework.boot
            spring-boot-starter-web
        org.springframework.boot
            spring-boot-starter-test
            testorg.springframework.boot
            spring-boot-starter-activemq
        org.springframework.boot
                spring-boot-maven-plugin
            

引入 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表示持久化

jmsTemplate.setConnectionFactory(activeMQConnectionFactory); jmsTemplate.setDefaultDestination(queue); // 此处可不设置默认，在发送消息时也可设置队列 jmsTemplate.setSessionAcknowledgeMode(4);// 客户端签收模式 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);
    }
}

结果显示：

SpringBoot整合ActiveMQ

标签：cto   发送   over   value   ima   class   tar   依赖   conf   

原文地址：https://www.cnblogs.com/Libbo/p/11547852.html