rabbitmq延时队列

package com.anyec.mq.springRabbit;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

public class DeathProducer {
	@Autowired
	static RabbitTemplate amqpTemplate;

	public static void main(String[] args) throws IOException, TimeoutException {
		RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean();

		factory.setHost("localhost");
		factory.setPort(5672);
		factory.setUsername("javamq");
		factory.setPassword("javamq");
		factory.setRequestedChannelMax(10000);
		factory.setVirtualHost("/rabbit");
//		com.rabbitmq.client.ConnectionFactory fo = factory.getObject();
		com.rabbitmq.client.ConnectionFactory cf = factory.getRabbitConnectionFactory();
		Channel channel = cf.newConnection().createChannel();
		Map<String, Object> headers = new HashMap<String, Object>();
		headers.put("x-dead-letter-exchange", "orderCreateDeathExchange");
		headers.put("x-message-ttl", 10000);
		
		
//		channel.exchangeDelete("orderCreateExchange");
//		channel.exchangeDelete("orderCreateDeathExchange");
//		
//		channel.exchangeDeclare("orderCreateExchange", "direct", true, false, null);
//		channel.exchangeDeclare("orderCreateDeathExchange", "direct", true, false, null);
//		
//		
//		channel.queueDelete("orderCreateQueue");
//		channel.queueDelete("orderCreateDeathQueue");
//		
//		
//		channel.queueDeclare("orderCreateQueue", true, false, false, headers);
//		channel.queueDeclare("orderCreateDeathQueue", true, false, false, new HashMap<>());
//		
//		channel.queueUnbind("orderCreateQueue","orderCreateExchange",  "order.create");
//		channel.queueUnbind("orderCreateDeathQueue", "orderCreateDeathExchange",  "order.create");
//		
//		channel.queueBind("orderCreateQueue","orderCreateExchange",  "order.create");
//		channel.queueBind("orderCreateDeathQueue", "orderCreateDeathExchange", "order.create");

		
		new Thread(new DConsumer(channel,11)).start();
		new Thread(new DConsumer(channel,12)).start();
		new Thread(new DConsumer(channel,13)).start();
		new Thread(new DConsumer(channel,14)).start();
		new Thread(new DConsumer(channel,15)).start();
		new Thread(new DConsumer(channel,21)).start();
		new Thread(new DConsumer(channel,22)).start();
		new Thread(new DConsumer(channel,23)).start();
		new Thread(new DConsumer(channel,24)).start();
		new Thread(new DConsumer(channel,25)).start();
		
		new Thread(new DConsumer(channel,31)).start();
		new Thread(new DConsumer(channel,32)).start();
		new Thread(new DConsumer(channel,33)).start();
		new Thread(new DConsumer(channel,34)).start();
		new Thread(new DConsumer(channel,35)).start();
		new Thread(new DConsumer(channel,41)).start();
		new Thread(new DConsumer(channel,42)).start();
		new Thread(new DConsumer(channel,43)).start();
		new Thread(new DConsumer(channel,44)).start();
		new Thread(new DConsumer(channel,45)).start();
//		for (int i = 0; i < 10000000; i++) {
//			BasicProperties props = new BasicProperties().builder().headers(headers).build();
//			channel.basicPublish("orderCreateExchange", "order.create", props, (i+"死刑队列10s").getBytes());
//			try {
//				Thread.sleep(1);
//			} catch (InterruptedException e) {
//				e.printStackTrace();
//			}
//		}
//		ConnectionFactory ff=new CachingConnectionFactory(cf);
//		amqpTemplate=new RabbitTemplate(ff);
//		amqpTemplate.sen
	}
}

class DConsumer implements Runnable {

	private Channel channel;
	int id;

	public DConsumer(Channel channel,Integer id) {
		super();
		this.channel = channel;
		this.id=id;
	
	}

	@Override
	public void run() {
		CancelCallback cancelCallback = new CancelCallback() {

			@Override
			public void handle(String consumerTag) throws IOException {
				System.out.println(consumerTag + " ");

			}
		};
		DeliverCallback deliverCallback = new DeliverCallback() {

			@Override
			public void handle(String consumerTag, Delivery message) throws IOException {
				System.out.println(id+" "+consumerTag + " " + new String(message.getBody())+message.getProperties().getMessageId());
				channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
				
			}
		};
		while (true) {

			try {
				channel.basicConsume("orderCreateDeathQueue", deliverCallback, cancelCallback);
			} catch (IOException e) {
				e.printStackTrace();
			}

		}
	}

}