/**
     * 推送延遲消息
     * @param topic 
     * @param body 
     * @param producerGroup 
     * @return boolean
     */
    public boolean sendMessage(String topic, String body, String producerGroup)
    {
        try
        {
            Message recordMsg = new Message(topic, body.getBytes());
            producer.setProducerGroup(producerGroup);
 
            //設(shè)置消息延遲級(jí)別，我這里設(shè)置14，對(duì)應(yīng)就是延時(shí)10分鐘
            // "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
            recordMsg.setDelayTimeLevel(14);
            // 發(fā)送消息到一個(gè)Broker
            SendResult sendResult = producer.send(recordMsg);
            // 通過(guò)sendResult返回消息是否成功送達(dá)
            log.info("發(fā)送延遲消息結(jié)果：======sendResult：{}", sendResult);
            DateFormat format =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            log.info("發(fā)送時(shí)間：{}", format.format(new Date()));
 
            return true;
        }
        catch (Exception e)
        {
            e.printStackTrace();
            log.error("延遲消息隊(duì)列推送消息異常:{},推送內(nèi)容:{}", e.getMessage(), body);
        }
        return false;
    }

/**
     * 接收延遲消息
     * 
     * @param topic
     * @param consumerGroup
     * @param messageHandler
     */
    public void messageListener(String topic, String consumerGroup, MessageListenerConcurrently messageHandler)
{
        ThreadPoolUtil.execute(() ->
        {
            try
            {
                DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
                consumer.setConsumerGroup(consumerGroup);
                consumer.setVipChannelEnabled(false);
                consumer.setNamesrvAddr(address);
                //設(shè)置消費(fèi)者拉取消息的策略，*表示消費(fèi)該topic下的所有消息，也可以指定tag進(jìn)行消息過(guò)濾
                consumer.subscribe(topic, "*");
                //消費(fèi)者端啟動(dòng)消息監(jiān)聽(tīng)，一旦生產(chǎn)者發(fā)送消息被監(jiān)聽(tīng)到，就打印消息，和rabbitmq中的handlerDelivery類(lèi)似
                consumer.registerMessageListener(messageHandler);
                consumer.start();
                log.info("啟動(dòng)延遲消息隊(duì)列監(jiān)聽(tīng)成功:" + topic);
            }
            catch (MQClientException e)
            {
                log.error("啟動(dòng)延遲消息隊(duì)列監(jiān)聽(tīng)失敗:{}", e.getErrorMessage());
                System.exit(1);
            }
        });
    }

/**
 * 延遲消息監(jiān)聽(tīng)
 * 
 */
@Component
public class CourseOrderTimeoutListener implements ApplicationListener<ApplicationReadyEvent>
{
 
    @Resource
    private MQUtil mqUtil;
 
    @Resource
    private CourseOrderTimeoutHandler courseOrderTimeoutHandler;
 
    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent)
{
        // 訂單超時(shí)監(jiān)聽(tīng)
        mqUtil.messageListener(EnumTopic.ORDER_TIMEOUT, EnumGroup.ORDER_TIMEOUT_GROUP, courseOrderTimeoutHandler);
    }
}

/**
 *  實(shí)現(xiàn)監(jiān)聽(tīng)
 */
@Slf4j
@Component
public class CourseOrderTimeoutHandler implements MessageListenerConcurrently
{
 
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg : list)
        {
            // 得到消息體
            String body = new String(msg.getBody());
            JSONObject userJson = JSONObject.parseObject(body);
            TCourseBuy courseBuyDetails = JSON.toJavaObject(userJson, TCourseBuy.class);
 
            // 處理具體的業(yè)務(wù)邏輯，，，，，
 
      DateFormat format =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
           log.info("消費(fèi)時(shí)間：{}", format.format(new Date()));
           
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

byte[] messageBodyBytes = "Hello, world!".getBytes();  
AMQP.BasicProperties properties = new AMQP.BasicProperties();  
properties.setExpiration("60000");  
channel.basicPublish("my-exchange", "queue-key", properties, messageBodyBytes);  

String msg = "hello word";  
MessageProperties messageProperties = newMessageProperties();  
messageProperties.setExpiration("6000");
messageProperties.setCorrelationId(UUID.randomUUID().toString().getBytes());
Message message = newMessage(msg.getBytes(), messageProperties);
rabbitTemplate.convertAndSend("delay", "delay",message);

package wang.raye.rabbitmq.demo1;
import org.springframework.amqp.core.AcknowledgeMode;  
import org.springframework.amqp.core.Binding;  
import org.springframework.amqp.core.BindingBuilder;  
import org.springframework.amqp.core.DirectExchange;  
import org.springframework.amqp.core.Message;  
import org.springframework.amqp.core.Queue;  
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;  
import org.springframework.amqp.rabbit.connection.ConnectionFactory;  
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;  
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;  
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;
@Configuration
publicclassDelayQueue{  
    /** 消息交換機(jī)的名字*/
    publicstaticfinalString EXCHANGE = "delay";
    /** 隊(duì)列key1*/
    publicstaticfinalString ROUTINGKEY1 = "delay";
    /** 隊(duì)列key2*/
    publicstaticfinalString ROUTINGKEY2 = "delay_key";
    /**
     * 配置鏈接信息
     * @return
     */
    @Bean
    publicConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = newCachingConnectionFactory("120.76.237.8",5672);
        connectionFactory.setUsername("kberp");
        connectionFactory.setPassword("kberp");
        connectionFactory.setVirtualHost("/");
        connectionFactory.setPublisherConfirms(true); // 必須要設(shè)置
        return connectionFactory;
    }
    /**  
     * 配置消息交換機(jī)
     * 針對(duì)消費(fèi)者配置  
        FanoutExchange: 將消息分發(fā)到所有的綁定隊(duì)列，無(wú)routingkey的概念  
        HeadersExchange ：通過(guò)添加屬性key-value匹配  
        DirectExchange:按照routingkey分發(fā)到指定隊(duì)列  
        TopicExchange:多關(guān)鍵字匹配  
     */  
    @Bean  
    publicDirectExchange defaultExchange() {  
        returnnewDirectExchange(EXCHANGE, true, false);
    }
    /**
     * 配置消息隊(duì)列2
     * 針對(duì)消費(fèi)者配置  
     * @return
     */
    @Bean
    publicQueue queue() {  
       returnnewQueue("delay_queue2", true); //隊(duì)列持久  
    }
    /**
     * 將消息隊(duì)列2與交換機(jī)綁定
     * 針對(duì)消費(fèi)者配置  
     * @return
     */
    @Bean  
    @Autowired
    publicBinding binding() {  
        returnBindingBuilder.bind(queue()).to(defaultExchange()).with(DelayQueue.ROUTINGKEY2);  
    }
    /**
     * 接受消息的監(jiān)聽(tīng)，這個(gè)監(jiān)聽(tīng)會(huì)接受消息隊(duì)列1的消息
     * 針對(duì)消費(fèi)者配置  
     * @return
     */
    @Bean  
    @Autowired
    publicSimpleMessageListenerContainer messageContainer2(ConnectionFactory connectionFactory) {  
        SimpleMessageListenerContainer container = newSimpleMessageListenerContainer(connectionFactory());  
        container.setQueues(queue());  
        container.setExposeListenerChannel(true);  
        container.setMaxConcurrentConsumers(1);  
        container.setConcurrentConsumers(1);  
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //設(shè)置確認(rèn)模式手工確認(rèn)  
        container.setMessageListener(newChannelAwareMessageListener() {
            publicvoid onMessage(Message message, com.rabbitmq.client.Channel channel) throwsException{
                byte[] body = message.getBody();  
                System.out.println("delay_queue2 收到消息 : "+ newString(body));  
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //確認(rèn)消息成功消費(fèi)  
            }  
        });  
        return container;  
    }  
}

package com.zjt.shop.config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
@Configuration
public class RedisListenerConfig {
    @Autowired
    private RedisTemplate redisTemplate;
    /**
     * @return
     */
    @Bean
    public RedisTemplate redisTemplateInit() {
        // key序列化
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        //val實(shí)例化
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        return redisTemplate;
    }
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }
}

package com.zjt.shop.common.util;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.zjt.shop.modules.order.service.OrderInfoService;
import com.zjt.shop.modules.product.entity.OrderInfoEntity;
import com.zjt.shop.modules.product.mapper.OrderInfoMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
    public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
        super(listenerContainer);
    }
    @Autowired
    private OrderInfoMapper orderInfoMapper;
    /**
     * 針對(duì)redis數(shù)據(jù)失效事件，進(jìn)行數(shù)據(jù)處理
     * @param message
     * @param pattern
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
      try {
          String key = message.toString();
          //從失效key中篩選代表訂單失效的key
          if (key != null && key.startsWith("order_")) {
              //截取訂單號(hào)，查詢(xún)訂單，如果是未支付狀態(tài)則為-取消訂單
              String orderNo = key.substring(6);
              QueryWrapper<OrderInfoEntity> queryWrapper = new QueryWrapper<>();
              queryWrapper.eq("order_no",orderNo);
              OrderInfoEntity orderInfo = orderInfoMapper.selectOne(queryWrapper);
              if (orderInfo != null) {
                  if (orderInfo.getOrderState() == 0) {   //待支付
                      orderInfo.setOrderState(4);         //已取消
                      orderInfoMapper.updateById(orderInfo);
                      log.info("訂單號(hào)為【" + orderNo + "】超時(shí)未支付-自動(dòng)修改為已取消狀態(tài)");
                  }
              }
          }
      } catch (Exception e) {
          e.printStackTrace();
          log.error("【修改支付訂單過(guò)期狀態(tài)異常】：" + e.getMessage());
      }
    }
}