Rabbitmq延迟队列实现定时任务的方法

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 = new MessageProperties(); 
  messageProperties.setExpiration("6000");
  messageProperties.setCorrelationId(UUID.randomUUID().toString().getBytes());
  Message message = new Message(msg.getBytes(), messageProperties);
  rabbitTemplate.convertAndSend("delay", "delay",message);

messageProperties.setExpiration("6000"); 

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
public class DelayQueue { 
 /** 消息交换机的名字*/
 public static final String EXCHANGE = "delay";
 /** 队列key1*/
 public static final String ROUTINGKEY1 = "delay";
 /** 队列key2*/
 public static final String ROUTINGKEY2 = "delay_key";

 /**
  * 配置链接信息
  * @return
  */
 @Bean
 public ConnectionFactory connectionFactory() {
  CachingConnectionFactory connectionFactory = new CachingConnectionFactory("120.76.237.8",5672);

  connectionFactory.setUsername("kberp");
  connectionFactory.setPassword("kberp");
  connectionFactory.setVirtualHost("/");
  connectionFactory.setPublisherConfirms(true); // 必须要设置
  return connectionFactory;
 }

 /** 
  * 配置消息交换机
  * 针对消费者配置 
  FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念 
  HeadersExchange ：通过添加属性key-value匹配 
  DirectExchange:按照routingkey分发到指定队列 
  TopicExchange:多关键字匹配 
  */ 
 @Bean 
 public DirectExchange defaultExchange() { 
  return new DirectExchange(EXCHANGE, true, false);
 } 

 /**
  * 配置消息队列2
  * 针对消费者配置 
  * @return
  */
 @Bean
 public Queue queue() { 
  return new Queue("delay_queue2", true); //队列持久 

 }
 /**
  * 将消息队列2与交换机绑定
  * 针对消费者配置 
  * @return
  */
 @Bean 
 @Autowired
 public Binding binding() { 
  return BindingBuilder.bind(queue()).to(defaultExchange()).with(DelayQueue.ROUTINGKEY2); 
 } 

 /**
  * 接受消息的监听，这个监听会接受消息队列1的消息
  * 针对消费者配置 
  * @return
  */
 @Bean 
 @Autowired
 public SimpleMessageListenerContainer messageContainer2(ConnectionFactory connectionFactory) { 
  SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory()); 
  container.setQueues(queue()); 
  container.setExposeListenerChannel(true); 
  container.setMaxConcurrentConsumers(1); 
  container.setConcurrentConsumers(1); 
  container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认 
  container.setMessageListener(new ChannelAwareMessageListener() {

   public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {
    byte[] body = message.getBody(); 
    System.out.println("delay_queue2 收到消息 : " + new String(body)); 
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费 
   } 
  }); 
  return container; 
 } 
}