参考demo：G:\学习资料\java\java学习文档\java技术文档\消息队列文档\rabbimitmq\cd-xufei-rabbiitmq-parent\cd-xufei-rabbitmq-01 => limit

生产者示例：

public class Producer {

    /**
     * 队列名称
     */
    private static final String QUEUE_NAME = "queue_limit_re";

    public static void main(String[] args) throws Exception {
        // 获取连接
        Connection connection = ConnectionUtils.getConnection();
        // 建立通道
        Channel channel = connection.createChannel();
        // 申明一个队列
        /**
         *第一个参数：队列的名字
         *第二个参数：是否持久化  ：这个数据 发送到队列去的时候 这个数据 是否要持久化到数据库中去
         *第三个参数：是否排外
         *           第一个意思：是否允许这个通道以外的 其他消费者来消费这个数据
         *           第二个意思：连接关闭之后  这个队列是否自动删除
         *第四个参数：是否自动删除
         *           表示的是 当最后一个连接退出的时候是否要删除这个队列
         *第五个参数：创建队列 付带的头的信息(TTL   死信队列)
         *
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 生产者 要向这个队列中  发送一条消息
        /**
         * 第一个参数：交换机的名字
         * 第二个参数：路由的key 因为这里是将数据 直接发送到队列中  所以这里的路由key直接就是 队列的名字
         * 第三个参数：发送消息的时候  附带的一些属性信息
         * 第四个参数：发送的这个消息体的内容
         */
        channel.basicPublish("", QUEUE_NAME, null, ("我是消费端限流与消息重回队列程序").getBytes());
    }
}

消费者示例：

public class Consumer1 {

    private static final String QUEUE_NAME = "queue_limit_re";

    public static void main(String[] args) throws Exception {
        // 第一步：获取连接
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 告诉他只只要我没有确定 你就不要给我 分配这个任务
        // 说白了 意思是一次性只是接受一个消息 没有确定之前  这里是不会给分配任务的
        /**
         * 这里实际上是用在了消费端限流
         * 第一个参数 表示的是限制 本身消息的大小  如果是 0那么表示的是 不受限制
         * 第二个参数 表示的意思是  一次性最多 接受的 消息的条目数
         * 第三个参数 表示的是 是否将上面的两个应用应用于 整个 通道 false表示的是 只是自己适用
         */
        channel.basicQos(0, 1, false); // 这句话只是用来限流的

        // 申明一个消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 执行了很多的逻辑

                // 结果逻辑出错了
                System.out.println("执行了....:" + new String(body));
                String isFalse = "1";// 如果当前这个消息执行出错了 需要将消息重新返回队列
                if ("1".equals(isFalse)) {
                    // 需要将消息重回队列
                    /**
                     * 第一个参数是当前这条消息的标记
                     * 第二个参数表示是否批量处理
                     * 第三个参数是是否让消息重回队列
                     */
                    // 就要让消息重回队列
                    channel.basicNack(envelope.getDeliveryTag(), false, true);
                } else {
                    // 消费消息的时候如果设置为手动应答需要下面的设置，这样才会保证业务处理和消息消费同时成功
                    /**
                     * 第一个参数 当前这条消息的标记
                     * 第二个参数 是否自动应答
                     */
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        /**
         * 第一个参数：队列的名字
         * 第二个参数：是否自动应答
         * 第三个参数是 消费者
         */
        // 接下来是不是要绑定这个消费者  到这个队列
        channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    }
}