springBoot整合ActiveMQ实现延时发现消息

小咪咪 2022-04-25 06:00 226阅读 0赞

# 修改 activemq.xml 配置文件 #

> 生产者提供两个发送消息的方法，一个是即时发送消息，一个是延时发送消息。延时发送消息需要手动修改activemq目录conf下的activemq.xml配置文件，开启延时，设置**schedulerSupport="true"，****然后重启activemq即可**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70][]

# pom 坐标 #

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

# yml 配置 #

spring:
      activemq:
        broker-url: tcp://xx.xxx.xxx.xxx:61616
        pool:
          enabled: false
          max-connections: 100
          #idle-timeout: 3000
        packages:
          trust-all: true

# avtiveMQ config 配置 #

@Service
    public class JmsProducer {
    	protected final Log logger = LogFactory.getLog(getClass());
    	
    	@Resource
    	private JmsMessagingTemplate jmsMessagingTemplate;
    
    
    	public void sendMsg(String destinationName,Object message)
    	{
    		logger.info("===============>>>>>发送queue消息"+message);
    		Destination destination=new ActiveMQQueue(destinationName);
    		jmsMessagingTemplate.convertAndSend(destination,message);
    	}
    	
    	public void delaySend(String queueName, Object o, Long time) {
            logger.info("===============>>>>>发送queue消息"+o+",延时"+(time/1000L/60L)+"分钟");
    
            //获取连接工厂
            ConnectionFactory connectionFactory = this.jmsMessagingTemplate.getConnectionFactory();
            try {
                //获取连接
                Connection connection = connectionFactory.createConnection();
                connection.start();
                //获取session
                Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                // 创建一个消息队列
                Destination destination = session.createQueue(queueName);
                MessageProducer producer = session.createProducer(destination);
                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
                ObjectMessage message = session.createObjectMessage((Serializable) o);
                //设置延迟时间
                message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
                //发送
                producer.send(message);
                session.commit();
                producer.close();
                session.close();
                connection.close();
            } catch (Exception e) {
                e.getMessage();
            }
        }
     
    }

#  生产者 #

@Autowired
     private JmsProducer jmsProducer;
     
    jmsProducer.delaySend("pushFiveMinutes", save.getData(), 5 * 60 * 1000L);

# 消费者 #

@Slf4j
    @Component
    public class JmsConsumer {
    
     
        @JmsListener(destination = "pushFiveMinutes")
        public void pushFiveMinutes(String orderNo) {
          省略。。。
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2w4NDgxNjg_size_16_color_FFFFFF_t_70]: /images/20220205/3fbdd44d1e8f4c51829a0e996488cff1.png