java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > springboot rabbitmq整合消息持久化存储

springboot rabbitmq整合rabbitmq之消息持久化存储问题

作者：weixin_43831204

这篇文章主要介绍了springboot rabbitmq整合rabbitmq之消息持久化存储问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教

rabbitmq消息持久化存储包含三个方面

exchange的持久化

在申明exchange的时候，有个参数：durable。

当该参数为true，则对该exchange做持久化，重启rabbitmq服务器，该exchange不会消失。

durable的默认值为true

public class DirectExchange extends AbstractExchange {
    public static final DirectExchange DEFAULT = new DirectExchange("");
    public DirectExchange(String name) {
        super(name);
    }
    public DirectExchange(String name, boolean durable, boolean autoDelete) {
        super(name, durable, autoDelete);
    }
    public DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        super(name, durable, autoDelete, arguments);
    }
    public final String getType() {
        return "direct";
    }
}
public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {
    private final String name;
    private final boolean durable;
    private final boolean autoDelete;
    private final Map<String, Object> arguments;
    private volatile boolean delayed;
    private boolean internal;
    public AbstractExchange(String name) {
        this(name, true, false);
    }
    public AbstractExchange(String name, boolean durable, boolean autoDelete) {
        this(name, durable, autoDelete, (Map)null);
    }
    public AbstractExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        this.name = name;
        this.durable = durable;
        this.autoDelete = autoDelete;
        if (arguments != null) {
            this.arguments = arguments;
        } else {
            this.arguments = new HashMap();
        }
    }

queue的持久化

申明队列时也有个参数：durable。

当该参数为true，则对该queue做持久化，重启rabbitmq服务器，该queue不会消失。

durable的默认值为true

public Queue(String name) {
        this(name, true, false, false);
    }
    public Queue(String name, boolean durable) {
        this(name, durable, false, false, (Map)null);
    }
    public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete) {
        this(name, durable, exclusive, autoDelete, (Map)null);
    }
    public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        Assert.notNull(name, "'name' cannot be null");
        this.name = name;
        this.actualName = StringUtils.hasText(name) ? name : Base64UrlNamingStrategy.DEFAULT.generateName() + "_awaiting_declaration";
        this.durable = durable;
        this.exclusive = exclusive;
        this.autoDelete = autoDelete;
        this.arguments = (Map)(arguments != null ? arguments : new HashMap());
    }

message的持久化

前面我们已经讲到exchange与queue的持久化，那么message如何持久化呢？

我们在使用rabbit-client做消息持久化时，设置了BasicProperties的deliveryMode为2，做消息的持久化。

AMQP.BasicProperties properties = new AMQP.BasicProperties.
                Builder().
                deliveryMode(2).
                build();
        channel.basicPublish("ex.pc", "key.pc",  properties, "hello world".getBytes());

那么整合了spring boot，使用RabbitTemplate如何做持久化？

首先，我们来到经常的使用的消息发送方法：RabbitTemplate类下的convertAndSend

@Override
    public void convertAndSend(String exchange, String routingKey, final Object object) throws AmqpException {
        convertAndSend(exchange, routingKey, object, (CorrelationData) null);
    }

然后调用了该类下的重载方法：convertAndSend。该方法中将object 转换成了message

@Override
    public void convertAndSend(String exchange, String routingKey, final Object object,
            @Nullable CorrelationData correlationData) throws AmqpException {
        send(exchange, routingKey, convertMessageIfNecessary(object), correlationData);
    }

在做消息转换的时候，我们注意到，传入了一个MessageProperties对象

protected Message convertMessageIfNecessary(final Object object) {
        if (object instanceof Message) {
            return (Message) object;
        }
        return getRequiredMessageConverter().toMessage(object, new MessageProperties());
    }

在MessageProperties中，有个deliveryMode属性，该属性默认值为：MessageDeliveryMode.PERSISTENT（持久化的）

 public MessageProperties() {
        this.deliveryMode = DEFAULT_DELIVERY_MODE;
        this.priority = DEFAULT_PRIORITY;
    }
static {
    DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT;
    DEFAULT_PRIORITY = 0;
}

消息转换完成后，调用时同类方法的send方法

@Override
    public void send(final String exchange, final String routingKey,
            final Message message, @Nullable final CorrelationData correlationData)
            throws AmqpException {
        execute(channel -> {
            doSend(channel, exchange, routingKey, message,
                    (RabbitTemplate.this.returnCallback != null
                            || (correlationData != null && StringUtils.hasText(correlationData.getId())))
                            && RabbitTemplate.this.mandatoryExpression.getValue(
                                    RabbitTemplate.this.evaluationContext, message, Boolean.class),
                    correlationData);
            return null;
        }, obtainTargetConnectionFactory(this.sendConnectionFactorySelectorExpression, message));
    }

该方法又调用了doSend方法

public void doSend(Channel channel, String exchangeArg, String routingKeyArg, Message message, // NOSONAR complexity
            boolean mandatory, @Nullable CorrelationData correlationData)
                    throws Exception { // NOSONAR TODO: change to IOException in 2.2.
        String exch = exchangeArg;
        String rKey = routingKeyArg;
        if (exch == null) {
            exch = this.exchange;
        }
        if (rKey == null) {
            rKey = this.routingKey;
        }
        if (logger.isDebugEnabled()) {
            logger.debug("Publishing message " + message
                    + "on exchange [" + exch + "], routingKey = [" + rKey + "]");
        }
        Message messageToUse = message;
        MessageProperties messageProperties = messageToUse.getMessageProperties();
        if (mandatory) {
            messageProperties.getHeaders().put(PublisherCallbackChannel.RETURN_LISTENER_CORRELATION_KEY, this.uuid);
        }
        if (this.beforePublishPostProcessors != null) {
            for (MessagePostProcessor processor : this.beforePublishPostProcessors) {
                messageToUse = processor.postProcessMessage(messageToUse, correlationData);
            }
        }
        setupConfirm(channel, messageToUse, correlationData);
        if (this.userIdExpression != null && messageProperties.getUserId() == null) {
            String userId = this.userIdExpression.getValue(this.evaluationContext, messageToUse, String.class);
            if (userId != null) {
                messageProperties.setUserId(userId);
            }
        }
        sendToRabbit(channel, exch, rKey, mandatory, messageToUse);
        // Check if commit needed
        if (isChannelLocallyTransacted(channel)) {
            // Transacted channel created by this template -> commit.
            RabbitUtils.commitIfNecessary(channel);
        }
    }

在该方法中我们终于看到了发送消息到rabbitmq的操作：sendToRabbit。

该方法将MessageProperties对象转换成了BasicProperties。

至此，我们终于了解了，spring rabbit 中如何实现messge的持久化。

默认的message就是持久化的

protected void sendToRabbit(Channel channel, String exchange, String routingKey, boolean mandatory,
            Message message) throws IOException {
        BasicProperties convertedMessageProperties = this.messagePropertiesConverter
                .fromMessageProperties(message.getMessageProperties(), this.encoding);
        channel.basicPublish(exchange, routingKey, mandatory, convertedMessageProperties, message.getBody());
    }

如何改变message的持久化属性？

根据上面的源码分析，spring中默认的message就是持久化的，如何改变持久化属性?

1、使用send方法，发送message。设置message中MessageProperties的属性deliveryMode

2、自定义MessageConverter，在消息转换时，设置MessageProperties的属性deliveryMode

3、自定MessagePropertiesConverter，在MessageProperties对象转换成BasicProperties时，设置deliveryMode

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:
阅读全文