生产者
/** * 消息被拒的情况 */ public class Produce0001 { private static final String NORMAL_EXCHANGE="normal_exchange"; public static void main(String[] args) throws Exception{ Channel channel = untils.getChannel(); channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); //该消息用作队列的个数限制 for(int i=0;i<10;i++) { String message="info"+i; channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes(StandardCharsets.UTF_8)); System.out.println("生产者发送消息"+message); } } }
消费者:
/** * 消息被拒的情况 */ public class Consumer0001 { //普通交换机 private static final String NORMAL_EXCHANGE="normal_exchange"; //死信交换机 private static final String DEAD_EXCHANGE="dead_exchange"; public static void main(String[] args) throws Exception{ Channel channel = untils.getChannel(); //声明死信交换机,类型为direct channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT); //声明死信队列 String deadQueue="dead_queue"; channel.queueDeclare(deadQueue,false,false,false,null); //死信队列绑定交换和routingKey值 channel.queueBind(deadQueue,DEAD_EXCHANGE,"lisi"); //正常队列绑定死信队列 Map<String,Object> params=new HashMap<>(); //正常队列设置死信交换机,参数key是固定值 params.put("x-dead-letter-exchange",DEAD_EXCHANGE); //正常队列设置死信routing-key,参数key是固定值 params.put("x-dead-letter-routing-key", "lisi"); //正常队列设置的最大限制长度 params.put("x-max-length",6); System.out.println("等待接收消息...."); String normalQueue="normal_queue"; channel.queueDeclare(normalQueue,false,false,false,params); channel.queueBind(normalQueue,NORMAL_EXCHANGE,"zhangsan"); DeliverCallback deliverCallback=(consumerTag, message) -> { String s = new String(message.getBody(), StandardCharsets.UTF_8); if (s.equals("info5")) { System.out.println("info5拒接"); channel.basicReject(message.getEnvelope().getDeliveryTag(), false); } else { System.out.println("01接收到消息"+s); channel.basicAck(message.getEnvelope().getDeliveryTag(), false); } }; channel.basicConsume(normalQueue,false,deliverCallback,consumerTag -> {}); } }
结果: