生产端
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory= new ConnectionFactory();
connectionFactory.setHost(“192.168.44.140”);
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");

    Connection connection = connectionFactory.newConnection();
    Channel channel= connection.createChannel();
    String exchange="test_return_exchange";
    String routingKey="test.return.test1";
    String msg="测试return";
    channel.addReturnListener(new ReturnListener() {
        public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
            System.out.println("replyCode="+replyCode);
            System.out.println("replyText="+replyText);
            System.out.println("exchange="+exchange);
            System.out.println("routingKey="+routingKey);
            System.out.println("properties="+properties);
            System.out.println("body="+new String(body));

        }
    });

    //mandatory  为true时无法路由的消息会被监听   可通过addReturnListener监听到
    channel.basicPublish(exchange,routingKey,true,null,msg.getBytes());
}

消费端
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
ConnectionFactory connectionFactory= new ConnectionFactory();
connectionFactory.setHost(“192.168.44.140”);
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");

    Connection connection = connectionFactory.newConnection();
    Channel channel= connection.createChannel();
    String exchange="test_return_exchange";
    String routingKey="test.return";

    channel.exchangeDeclare(exchange,"topic",true,false,null);
    channel.queueDeclare("test_return_queue",true,false,false,null);
    channel.queueBind("test_return_queue",exchange,routingKey);

    QueueingConsumer queueingConsumer= new QueueingConsumer(channel);
    channel.basicConsume("test_return_queue",true,queueingConsumer);

    while(true){
        QueueingConsumer.Delivery delivery= queueingConsumer.nextDelivery();
        String msg = new String(delivery.getBody());
        System.err.println("消费端消费消息"+msg);
    }

}

控制台