RabbitMQ的基础知识

目录

• RabbitMQ
  • 1.对MQ的介绍
  • 2.RabbitMQ的六种模式 及工作原理
  • 3.hello world队列
  • 4.工作队列模式
  • 5.消息应答机制
    • 自动应答
    • 手动应答
    • 消息自动进行重新入队
  • 6.RabbitMQ的持久化,不公平分发及预取值
  • 7.发布确认
  • 8.交换机
    • <1>交换机的认识
    • <2>交换机具体介绍
  • 9.死信队列
    • <1>认识死信队列
    • <2>死信实战

<!--rabbitmq 依赖客户端-->
<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.8.0</version>
</dependency>

public class Producer {
    //建立队列
    private static final String QUEUE_NAME="hello";
    public static void main(String[] args)  {
        //创建连接工场
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setUsername("guest");
        factory.setUsername("guest");

        try {
            //建立连接和信道
            //channel 实现了自动 close 接口 自动关闭 不需要显示关闭
            Connection connection=factory.newConnection();
            Channel channel=connection.createChannel();

            /**
             * 生成一个队列,并将信道和队列连接
             * 1.队列名称
             * 2.队列里面的消息是否持久化 默认消息存储在内存中
             * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
             * 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
             * 5.其他参数
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            String message="hello world";
            /**
             * 发送一个消息
             * 1.发送到那个交换机
             * 2.路由的 key 是哪个
             * 3.其他的参数信息
             * 4.发送消息的消息体
             */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("消息发送成功");
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

public class Consumer {
    //定义队列名
    private static final String QUEUE_NAME="hello";

    public static void main(String[] args) {
        //建立连接和信道
        try {
            ConnectionFactory factory=new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection=factory.newConnection();
            Channel channel=connection.createChannel();
            System.out.println("等待接收消息");


            /**
             *1.同一个会话， consumerTag 是固定的 可以做此会话的名字， deliveryTag 每次接收消息+1，可以做此消息处理通道的名字。
             *2.包含消息的字节形式的类
             */
            DeliverCallback deliverCallback=(consumerTag,delivery)->{
                String message=new String(delivery.getBody());
                System.out.println(message);
            };
            CancelCallback cancelCallback=(consumerTag)->{
                System.out.println("消息消费被取消");
            };
            /* 消费者消费消息
             * 1.消费哪个队列
             * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
             * 3.消费者未成功消费的回调
             * 4.消费者取消消费的回调
             */
            channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);



        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

public class Producer {
    public static void main(String[] args) throws IOException, InterruptedException {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("失败");
            return;
        }
        channel.queueDeclare(RabbitMQChannelUtil.QUEUE_NAME,false,false,false,null);
        int i=0;
        while (true){
            String message="消息"+i;
            i++;
            /**
             * 发送一个消息
             * 1.发送到那个交换机
             * 2.路由的 key 是哪个
             * 3.其他的参数信息
             * 4.发送消息的消息体
             */
            channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME,null,message.getBytes());
            System.out.println(message);
            Thread.sleep(500);
        }
    }
}

public class Consumer {
    public static void main(String[] args) {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("消费失败");
            return;
        }
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message=new String(delivery.getBody());
            System.out.println(Thread.currentThread().getName()+"消费了"+message);
        };
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("消息消费被取消");
        };
        Thread[] threads=new Thread[5];
        for (int i = 0; i <threads.length ; i++) {
            threads[i]=new Thread(()->{
                try {
                    System.out.println(Thread.currentThread().getName()+"启动等待消费");
                    channel.basicConsume(RabbitMQChannelUtil.QUEUE_NAME,true,deliverCallback,cancelCallback);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
        for (int i = 0; i <threads.length ; i++) {
            threads[i].start();
        }
    }
}

/* 消费者消费消息
 * 1.消费哪个队列
 * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
 * 3.当一个消息发送过来后的回调接口
 * 4.消费者取消消费的回调
 */
boolean ack=false;
channel.basicConsume(QUEUE_NAME,ack,deliverCallback,cancelCallback);

/**
 * 消息发送过来后的回调接口
 *1.同一个会话， consumerTag 是固定的 可以做此会话的名字， deliveryTag 每次接收消息+1，可以做此消息处理通道的名字。
 *2.消息类
 */
DeliverCallback deliverCallback=(consumerTag,delivery)->{
    String message=new String(delivery.getBody());
    System.out.println(message);
    /**
     * 参数说明
     * 1.消息的标记tag
     * 2.是否批量应答
     */
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
};

channel.queueDeclare(RabbitMQChannelUtil.QUEUE_NAME,true,false,false,null);

channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME,null,message.getBytes());

channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

channel.basicQos(1);

channel.confirmSelect();

public static void singleConfirm(){
    try {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("信道建立失败");
            return;
        }
        //开启发布确认
        channel.confirmSelect();
        long begin=System.currentTimeMillis();
        for (int i = 0; i <MESSAGE_COUNT ; i++) {
            String message=i+"";
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            //可以加时间参数，当消息发送失败或超过参数时间没成功，则返回false
            boolean flag=channel.waitForConfirms();
            //如果失败可以重发
            if(flag){
                System.out.println(message+"发送成功");
            }else {
                //这里可以实现重发
                System.out.println(message+"发送失败");
            }
        }
        long end=System.currentTimeMillis();
        System.out.println("发送"+MESSAGE_COUNT+"条消息，耗时"+(end-begin)+"ms");
    }catch (Exception e){
        e.printStackTrace();
    }
}

public static void batchConfirm(){
    try {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("建立连接失败");
            return;
        }
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //当100条消息发布成功时，再确认
        int ackMessageCount=100;
        //未确认的消息个数
        int needAckMessageCount=0;
        //开启发布确认
        channel.confirmSelect();
        long begin=System.currentTimeMillis();
        for (int i = 0; i <MESSAGE_COUNT ; i++) {
            String message=i+"";
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            needAckMessageCount++;
            if(needAckMessageCount == ackMessageCount){
                //确认
                channel.waitForConfirms();
                needAckMessageCount=0;
            }
        }
        //判断可能还有消息未发送，再发送依次
        if(needAckMessageCount > 0){
            channel.waitForConfirms();
        }
        long end= System.currentTimeMillis();
        System.out.println("发送"+MESSAGE_COUNT+"条消息，耗时"+(end-begin)+"ms");
    }catch (Exception e){
        e.printStackTrace();
    }
}

public static void asyncConfirm() throws Exception {
    try (Channel channel = RabbitMQChannelUtil.getChannel()) {
        if(channel == null){
            return;
        }
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new
                ConcurrentSkipListMap<>();
        /**
         * 确认收到消息的一个回调
         * 1.消息序列号
         * 2.true 可以确认小于等于当前序列号的消息
         * false 确认当前序列号消息
         */
        ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
            if (multiple) {
                //返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed =
                        outstandingConfirms.headMap(sequenceNumber, true);
                //清除该部分未确认消息
                confirmed.clear();
            }else{
                //只清除当前序列号的消息
                outstandingConfirms.remove(sequenceNumber);
            }
        };
        ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
            String message = outstandingConfirms.get(sequenceNumber);
            System.out.println("发布的消息"+message+"未被确认，序列号"+sequenceNumber);
        };
        /**
         * 添加一个异步确认的监听器
         * 1.确认收到消息的回调
         * 2.未收到消息的回调
         */
        channel.addConfirmListener(ackCallback, null);
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            /**
             * channel.getNextPublishSeqNo()获取下一个消息的序列号
             * 通过序列号与消息体进行一个关联
             * 全部都是未确认的消息体
             */
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        }
        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) +
                "ms");
    } 
}

channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

String queueName = channel.queueDeclare().getQueue();

String queueName = channel.queueDeclare().getQueue();

//声明交换机名称及类型
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串
 channel.queueBind(queueName, EXCHANGE_NAME, "");

public class Producer {
    private static final String NORMAL_EXCHANGE="normal_exchange";
    public static void main(String[] args) {
        try {
            Channel channel= RabbitMQChannelUtil.getChannel();
            if(channel == null){
                return;
            }
            //声明交换机类型
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            //设置消息TTL时间
            AMQP.BasicProperties basicProperties=new AMQP.BasicProperties().builder().expiration("1000").build();
            //用作演示消息队列的限制个数
            for (int i = 0; i <10 ; i++) {
                String message="info"+i;
                channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes());
                System.out.println("生产者发送消息");
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

public class NormalConsumer {
    //普通交换机名称
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";
    public static void main(String[] argv) throws Exception {
        Channel channel = RabbitMQChannelUtil.getChannel();
        if(channel == null){
            return;
        }
        //声明死信和普通交换机 类型为 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");

        String normalQueue = "normal-queue";
        channel.queueDeclare(normalQueue, false, false, false, params);
        channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
        System.out.println("等待接收消息.....");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("NormalConsumer 接收到消息"+message);
        };
        channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
        });
    }
}

public class DeadConsumer {
 private static final String DEAD_EXCHANGE = "dead_exchange";
 public static void main(String[] argv) throws Exception {
        Channel channel = RabbitMQChannelUtil.getChannel();
        if (channel == null) {
         return;
        }
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
        System.out.println("等待接收死信队列消息.....");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
         String message = new String(delivery.getBody(), "UTF-8");
         System.out.println("DeadConsumer 接收死信队列的消息" + message);
        };
        channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
        });
 }