//設(shè)置隊列上所有的消息的有效期,單位為毫秒
Map<String, Object> argss = new HashMap<String , Object>();
arguments.put("x-message-ttl " , 5000);//5秒鐘
channel.queueDeclare(queueName , durable , exclusive , autoDelete , arguments) ;

rabbitmqctl set_policy TTL ".*" '{"message-ttl":60000}' --apply-to queues

$ curl -i -u guest:guest -H "content-type:application/json"  -XPUT -d'{"auto_delete":false,"durable":true,"arguments":{"x-message-ttl": 60000}}' 
http://ip:15672/api/queues/{vhost}/{queuename}

Builder bd = new AMQP.BasicProperties().builder();
bd.deliveryMode(2);//持久化
bd.expiration("100000");//設(shè)置消息有效期100秒鐘
BasicProperties pros = bd.build();
String message = "測試ttl消息";
channel.basicPublish(EXCHANGE_NAME, "error", true,false, pros, message.getBytes());

$ curl -i -u guest:guest -H "content-type:application/json"  -XPOST -d
'{"properties":{"expiration":"60000"},"routing_key":"routingkey","payload":"my body","payload_encoding":"string"}'  
http://localhost:15672/api/exchanges/{vhost}/{exchangename}/publish

package cn.wkp.rabbitmq.newest.ttl;
 
import java.util.HashMap;
import java.util.Map;
 
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.BasicProperties.Builder;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
 
import cn.wkp.rabbitmq.util.ConnectionUtil;
 
/**
 * 
 * @ClassName: Send
 * @Description: 消息有效期 
 * @author wkg
 * @date: 2019年4月1日 下午11:28:22
 */
public class Send {
 
	private final static String EXCHANGE_NAME = "ttl_exchange";
	private final static String QUEUE_NAME = "ttl_queue";
 
	public static void main(String[] argv) throws Exception {
		// 獲取到連接以及mq通道
		Connection connection = ConnectionUtil.getConnection();
		// 從連接中創(chuàng)建通道
		Channel channel = connection.createChannel();
 
		// 聲明交換機
		channel.exchangeDeclare(EXCHANGE_NAME, "direct",true);
		
		//*****1:通過隊列設(shè)置有效期 2:通過消息屬性設(shè)置有效期,如果都設(shè)置了以較小的為準*****
		//聲明隊列
		Map<String, Object> arguments=new HashMap<String,Object>();
		//設(shè)置隊列上所有的消息的有效期,單位為毫秒
		arguments.put("x-message-ttl", 5000);//5秒鐘
		channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);
		//綁定
		channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
		
		Builder bd = new AMQP.BasicProperties().builder();
		bd.deliveryMode(2);//持久化
		bd.expiration("100000");//設(shè)置消息有效期100秒鐘
		BasicProperties pros = bd.build();
		String message = "測試ttl消息";
		channel.basicPublish(EXCHANGE_NAME, "error", true,false, pros, message.getBytes());
		System.out.println("Sent message:" + message);
//		 關(guān)閉通道和連接
		channel.close();
		connection.close();
	}
}

Map<String, Object> args = new HashMap<String, Object>();  
args.put("x-expires", 18000);  //隊列有效期18秒
channel.queueDeclare("myqueue", false, false, false, args);  

channel.exchangeDeclare("dlx_exchange" , "direct"); //創(chuàng)建DLX: dlx_exchange
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange" , "dlx_exchange ");//設(shè)置DLX
args.put("x-dead-letter-routing-key" , "dlx-routing-key");//設(shè)置DLX的路由鍵(可以不設(shè)置)
//為隊列myqueue 添加DLX
channel.queueDeclare("myqueue" , false , false , false , args);

package cn.wkp.rabbitmq.newest.dlx;
 
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
 
import cn.wkp.rabbitmq.util.ConnectionUtil;
 
public class SendDLX {
 
	public static void main(String[] args) throws Exception {
		Connection connection = ConnectionUtil.getConnection();
		Channel channel = connection.createChannel();
		//聲明一個交換機，做死信交換機用
		channel.exchangeDeclare("dlx_exchange", "topic", true, false, null);
		//聲明一個隊列，做死信隊列用
		channel.queueDeclare("dlx_queue", true, false, false, null);
		//隊列綁定到交換機上
		channel.queueBind("dlx_queue", "dlx_exchange", "dlx.*");
		
		channel.exchangeDeclare("normal_exchange", "fanout", true, false, null);
		Map<String, Object> arguments=new HashMap<String, Object>();
		arguments.put("x-message-ttl" , 5000);//設(shè)置消息有效期1秒，過期后變成私信消息，然后進入DLX
		arguments.put("x-dead-letter-exchange" , "dlx_exchange");//設(shè)置DLX
		arguments.put("x-dead-letter-routing-key" , "dlx.test");//設(shè)置DLX的路由鍵(可以不設(shè)置)
		//為隊列normal_queue 添加DLX
		channel.queueDeclare("normal_queue", true, false, false, arguments);
		channel.queueBind("normal_queue", "normal_exchange", "");
		
		channel.basicPublish("normal_exchange", "", MessageProperties.PERSISTENT_TEXT_PLAIN, ("測試死信消息").getBytes());
		System.out.println("發(fā)送消息時間:"+ConnectionUtil.formatDate(new Date()));
		
		channel.close();
		connection.close();
	}
}

package cn.wkp.rabbitmq.newest.dlx;
 
import java.io.IOException;
import java.util.Date;
 
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
 
import cn.wkp.rabbitmq.util.ConnectionUtil;
 
public class RecvDLX {
 
	public static void main(String[] argv) throws Exception {
		Connection connection = ConnectionUtil.getConnection();
		final Channel channel = connection.createChannel();
 
		channel.exchangeDeclare("dlx_exchange", "topic", true, false, null);
		channel.queueDeclare("dlx_queue", true, false, false, null);
		channel.queueBind("dlx_queue", "dlx_exchange", "dlx.*");
 
		// 指該消費者在接收到隊列里的消息但沒有返回確認結(jié)果之前,它不會將新的消息分發(fā)給它。
		channel.basicQos(1);
 
		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
					throws IOException {
				System.out.println("消費者收到消息:" + new String(body)+",當前時間:"+ConnectionUtil.formatDate(new Date()));
				// 消費者手動發(fā)送ack應(yīng)答
				channel.basicAck(envelope.getDeliveryTag(), false);
			}
		};
		System.out.println("消費死信隊列中的消息======================");
		// 監(jiān)聽隊列
		channel.basicConsume("dlx_queue", false, consumer);
	}
}