RabbitMq数据发送监听

前言

在工作中使用rabbit传输数据时，可能会因为数据、网络等问题，导致数据发送或者接收失败
如果对此类问题没有做好处理，就会存在丢失数据的问题，为此，引入了ConfirmCallback与ReturnCallback，来保证系统能够做到更好的数据监听、以及消费失败的数据做好相应的补偿；

ConfirmCallback与ReturnCallback也被称为Rabbitmq的消息确认机制；

这两个方法主要解决是否发送到交换机和是否发送到队列的监控问题

实现

@Slf4j
@Configuration
public class RabbitCallBackConfig {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Value(value = "${spring.profiles.active}")
    private String activeProfile;

    @SneakyThrows
    @PostConstruct
    public void enableConfirmCallBack() {
        //confirm 监听，当消息成功发到交换机 ack = true，没有发送到交换机 ack = false
        //correlationData 可在发送时指定消息唯一 id
        rabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {
            if (!ack) {
                // 消息没有发送到交换机，则发送邮件通知
                rabbitTemplate.send(MessageQueueContant.MESSAGE_EXCHANGE + "_" + activeProfile,
                        MessageQueueContant.SEND_MAIL_QUEUE_KEY,
                        getMessage("【重要】 消息未发送到交换机", "", null));
            }
        }));

        //当消息成功发送到交换机没有路由到队列触发此监听
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) ->
                rabbitTemplate.send(MessageQueueContant.MESSAGE_EXCHANGE + "_" + activeProfile,
                        MessageQueueContant.SEND_MAIL_QUEUE_KEY,
                        getMessage(String.format("【重要】 消息未路由到队列，交换机名称：%s", exchange),
                                new String(message.getBody(), StandardCharsets.UTF_8), message)));
    }

    @SneakyThrows
    private Message getMessage(String title, String content, Message message) {
        MailModel mailModel = new MailModel();
        mailModel.setAddressee(Arrays.asList("yangjiawen@dstcar.com", "fengtianzhu@dstcar.com", "dailei@dstcar.com"));
        mailModel.setTitle(Objects.toString(title, "【重要】队列发送消息失败 -- " + this.activeProfile));
        mailModel.setText(content);
        return new Message(JsonUtil.writeValueAsBytes(mailModel), getMessageProperties(message));
    }

    private MessageProperties getMessageProperties(Message message) {
        MessageProperties messageProperties = Objects.isNull(message) ?
                new MessageProperties() : message.getMessageProperties();
        messageProperties.setContentType("application/json");
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        return messageProperties;
    }

}

关联技术

  rabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {
            if (!ack) {
                // 消息没有发送到交换机，则发送邮件通知
                rabbitTemplate.send(MessageQueueContant.MESSAGE_EXCHANGE + "_" + activeProfile,
                        MessageQueueContant.SEND_MAIL_QUEUE_KEY,
                        getMessage("【重要】 消息未发送到交换机", "", null));
            }
        }));

这段代码使用了一个技术，就是方法的入参是函数式接口，如果不熟悉java8看起来这段代码多少有些疑惑，我们来看下这个方法的入参

    public void setConfirmCallback(ConfirmCallback confirmCallback) {
        Assert.state(this.confirmCallback == null || this.confirmCallback == confirmCallback,
                "Only one ConfirmCallback is supported by each RabbitTemplate");
        this.confirmCallback = confirmCallback;
    }

入参是ConfirmCallback类，我们来看下这个类

@FunctionalInterface
    public interface ConfirmCallback {

        /**
         * Confirmation callback.
         * @param correlationData correlation data for the callback.
         * @param ack true for ack, false for nack
         * @param cause An optional cause, for nack, when available, otherwise null.
         */
        void confirm(@Nullable CorrelationData correlationData, boolean ack, @Nullable String cause);

    }

可以看到他是一个接口类，但是加了@FunctionalInterface注解来修饰，说明他是一个函数式接口，那函数式接口有什么作用呢，看了下面的代码基本就知道了。
我定义一个函数式接口

@FunctionalInterface
public interface FunctionInterface {

    public void sendInfo(String aa,String bb);
}

我们调用一下他

public class TestInteface {

    public static void main(String[] args) {
        TestEntity ss = new TestEntity();
        ss.setTeeee((aa, bb)->{
            System.out.println(aa);
        });
    }

    static class TestEntity{
        public void setTeeee(FunctionInterface functionInterface){
            functionInterface.sendInfo("2","4");
        }
    }


}

这时候控制台输出式 2，我们可以看到函数式接口有一个很不一样的地方，就是当把他作为参数时，我们可以在传参时给他写实现，这样就可以增强一个接口的灵活性了。
其实java8有一个我们熟悉的类，专门处理这种情况，就是Consumer类

@FunctionalInterface
public interface Consumer<T> {

    /**
     * Performs this operation on the given argument.
     *
     * @param t the input argument
     */
    void accept(T t);


    default Consumer<T> andThen(Consumer<? super T> after) {
        Objects.requireNonNull(after);
        return (T t) -> { accept(t); after.accept(t); };
    }
}

这种特性能帮我们实现什么呢？其实有一种我们很常见的场景使用函数式接口作为参数就有很大的用处

    /**
     * 环绕增强
     */
    public void convert(String data, Consumer extendFunctionBefore, Consumer extendFunctionAfter) {
        Optional.ofNullable(extendFunctionBefore).ifPresent(s -> s.accept(1));
        System.out.println(data);
        Optional.ofNullable(extendFunctionAfter).ifPresent(s -> s.accept(1));
    }

这么处理我们就可以在我们想执行的方法前后灵活加上我们想执行的逻辑

结语

本来只是想说一下mq的，但是随着对实现的好奇，就一点点的说到了java8，技术总是一环套一环。