三天吃透 RabbitMQ 面试八股文

spring:    rabbitmq:           ##开启 confirm 确认机制        publisher-confirms: true

// 消息是否成功发送到Exchangefinal RabbitTemplate.ConfirmCallback confirmCallback = (CorrelationData correlationData, boolean ack, String cause) -> {            log.info("correlationData: " + correlationData);            log.info("ack: " + ack);            if(!ack) {                log.info("异常处理....");            }    };
rabbitTemplate.setConfirmCallback(confirmCallback);

spring:    rabbitmq:        ##触发ReturnCallback必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发ReturnCallback        template.mandatory: true

    final RabbitTemplate.ReturnCallback returnCallback = (Message message, int replyCode, String replyText, String exchange, String routingKey) ->            log.info("return exchange: " + exchange + ", routingKey: "                    + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);rabbitTemplate.setReturnCallback(returnCallback);

##设置消费端手动 ackspring.rabbitmq.listener.simple.acknowledge-mode=manual

    @RabbitListener(queues = RabbitMqConfig.MAIL_QUEUE)    public void onMessage(Message message, Channel channel) throws IOException {
        try {            Thread.sleep(5000);        } catch (InterruptedException e) {            e.printStackTrace();        }        long deliveryTag = message.getMessageProperties().getDeliveryTag();        //手工ack；第二个参数是multiple，设置为true，表示deliveryTag序列号之前（包括自身）的消息都已经收到，设为false则表示收到一条消息        channel.basicAck(deliveryTag, true);        System.out.println("mail listener receive: " + new String(message.getBody()));    }

##在单个请求中处理的消息个数，unack的最大数量spring.rabbitmq.listener.simple.prefetch=2

//单条消息大小限制，0代表不限制//global：限制限流功能是channel级别的还是consumer级别。当设置为false，consumer级别，限流功能生效，设置为true没有了限流功能，因为channel级别尚未实现。void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

  @Bean    public DirectExchange dlxExchange() {        return new DirectExchange(RabbitMqConfig.DLX_EXCHANGE);    }
    @Bean    public Queue dlxQueue() {        return new Queue(RabbitMqConfig.DLX_QUEUE, true);    }
    @Bean    public Binding bindingDeadExchange(Queue dlxQueue, DirectExchange deadExchange) {        return BindingBuilder.bind(dlxQueue).to(deadExchange).with(RabbitMqConfig.DLX_QUEUE);    }

    @Bean    public Queue sendSmsQueue() {        Map<String,Object> arguments = new HashMap<>(2);        // 绑定该队列到私信交换机        arguments.put("x-dead-letter-exchange", RabbitMqConfig.DLX_EXCHANGE);        arguments.put("x-dead-letter-routing-key", RabbitMqConfig.DLX_QUEUE);        return new Queue(RabbitMqConfig.MAIL_QUEUE, true, false, false, arguments);    }

@Component@Slf4jpublic class MQProducer {
    @Autowired    RabbitTemplate rabbitTemplate;
    @Autowired    RandomUtil randomUtil;
    @Autowired    UserService userService;
    final RabbitTemplate.ConfirmCallback confirmCallback = (CorrelationData correlationData, boolean ack, String cause) -> {            log.info("correlationData: " + correlationData);            log.info("ack: " + ack);            if(!ack) {                log.info("异常处理....");            }    };

    final RabbitTemplate.ReturnCallback returnCallback = (Message message, int replyCode, String replyText, String exchange, String routingKey) ->            log.info("return exchange: " + exchange + ", routingKey: "                    + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
    public void sendMail(String mail) {        //貌似线程不安全 范围100000 - 999999        Integer random = randomUtil.nextInt(100000, 999999);        Map<String, String> map = new HashMap<>(2);        String code = random.toString();        map.put("mail", mail);        map.put("code", code);
        MessageProperties mp = new MessageProperties();        //在生产环境中这里不用Message，而是使用 fastJson 等工具将对象转换为 json 格式发送        Message msg = new Message("tyson".getBytes(), mp);        msg.getMessageProperties().setExpiration("3000");        //如果消费端要设置为手工 ACK ，那么生产端发送消息的时候一定发送 correlationData ，并且全局唯一，用以唯一标识消息。        CorrelationData correlationData = new CorrelationData("1234567890"+new Date());
        rabbitTemplate.setMandatory(true);        rabbitTemplate.setConfirmCallback(confirmCallback);        rabbitTemplate.setReturnCallback(returnCallback);        rabbitTemplate.convertAndSend(RabbitMqConfig.MAIL_QUEUE, msg, correlationData);
        //存入redis        userService.updateMailSendState(mail, code, MailConfig.MAIL_STATE_WAIT);    }}

@Slf4j@Componentpublic class DeadListener {
    @RabbitListener(queues = RabbitMqConfig.DLX_QUEUE)    public void onMessage(Message message, Channel channel) throws IOException {
        try {            Thread.sleep(5000);        } catch (InterruptedException e) {            e.printStackTrace();        }        long deliveryTag = message.getMessageProperties().getDeliveryTag();        //手工ack        channel.basicAck(deliveryTag,false);        System.out.println("receive--1: " + new String(message.getBody()));    }}

GetResponse response = channel.basicGet(QUEUE_NAME, false);System.out.println(new String(response.getBody()));channel.basicAck(response.getEnvelope().getDeliveryTag(),false);

Message msg = new Message("tyson".getBytes(), mp);msg.getMessageProperties().setExpiration("3000");