Rabbitmq是如何来保证事务的

前言:

Rabbitmq舍弃了繁重的事务消息而使用了消息确认机制实现了分布式事务,实在是解耦之一大神器。其整个过程可以按照生产者、消费者分为两部分

1 发送确认

发送确认用来确保消息是否已送达消息队列。消息一旦到达消息服务,就会触发确认机制,可以分为两种情况:

  1. 对于无法被路由的消息,一旦没法找到一个队列来消费它,就会触发确认无法消费,此时ack=false。旦有一种情况例外,就是连exchange都没法找到,如果设置了mandatory, 此时就会先触发basic.return,就是会先触发returncallback回调。
  2. 对于可以被路由的消息,当消息被(所有的?)queue接受时,会触发ack=true;对于设置了持久化(persistent)的消息,当消息成功的持久化到硬盘上才会触发;对于设置了镜像(mirror)的消息,那么是当所有的mirror接受到这个消息。

2 消费确认(Delivery Acknowledgements)

消费确认用来确保消费者是否成功的消费了消息。一旦有消费者成功注册到相应的消息服务,消息将会被消息服务通过basic.deliver推(push)给消费者,此时消息会包含一个deliver tag用来唯一的标识消息。如果此时是手动模式,就需要手动的确认消息已经被成功消费,否则消息服务将会重发消息(因为消息已经持久化到了硬盘上,所以无论消息服务是不是可能挂掉,都会重发消息)。而且必须确认,无论是成功或者失败,否则会引起非常严重的问题

3 死信交换机(Dead Letter Exchanges)

有三种情况可能进死信交换机

  1. 被reject或者nack,并且requeue设置为false
  2. 消息最大存活时间(TTL)超时
  3. 消息数量超过最大队列长度

只需要设置一个args,就ok拉

channel.exchangeDeclare("some.exchange.name", "direct");
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange", "some.exchange.name");
channel.queueDeclare("myqueue", false, false, false, args);

4 参数和配置

Channel Prefetch Setting (QoS)

当前队列中未应答消息的数目,如果超过了,队列中将不再接受新的消息。加大这个值会增加消息的发送速度(Throughput),但是会加重消息队列的内存,所以100-300之间是一个比较理想的状态。可参考:http://next.rabbitmq.com/conf... :Channel Prefetch Setting (QoS)。

Consumer Prefetch

Qos有一个问题,一个channel是会被多个queue消费的,所以必须计算出所有queue中未应答的数目,然后给channel,这显然是非常不合理的,而且很麻烦,所以干脆还不如设置每个队列的数目更好

例子:

// 1. 一下子设置所有的queue都是10条unacknowledged
Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue", false, consumer);

//2. 分别设置10条
Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

//3. 分别设置channel和consume,个人不推荐
Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10, false); // Per consumer limit
channel.basicQos(15, true);  // Per channel limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

delivery tags

通道(channel)中的消息标志,按照正数递增,消息队列中用来标识消息的唯一标识

Blocked Connection Notifications

当消息服务器资源不足时,会向所有的生产者发送这个消息,我们可以捕获这个消息并做处理,资源可以是内存不足,cpu负载过重等等。

ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
connection.addBlockedListener(new BlockedListener() {
    public void handleBlocked(String reason) throws IOException {
        // Connection is now blocked
    }

    public void handleUnblocked() throws IOException {
        // Connection is now unblocked
    }
});

话说真出现block了也应该是在管理台直接给出警告,不过也可以做一下避免异常

Multiple

全应答标识。如果设置为true,一条消息应答了,那么之前的全部消息将被应答。比如目前channel中有delivery tags为5,6,7,8的消息,那么一旦8被应答,那么5,6,7将都被应答,如果设置为false,那么5,6,7将不会被应答。(不建议设置,毕竟一个channel中会绑定好多consumer)

basic.nack

当消息服务发生异常时,不会发送basic.ack,反而会发送一个basic.nack,而且不会自动requeue,此时需要消息发送方手动处理,进行重发。只有一种情况会发送nack:“basic.nack will only be delivered if an internal error occurs in the Erlang process responsible for a queue”

1 mq配置的listener到什么地方,是配置到每个微服务、或者是配置到mq中?如果在每个服务里面都配置concurency,是不是随着节点的增加,listener数量也会无限增加?2 ssl和max-queue-length的配置

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在,都有一个你没有努力的曾经。
公众号推荐
   一个历史类的公众号,欢迎关注
一两拨千金