rabbitmq-小结_rabbitlistener fanout

一、基础概念

rabbitmq是amqp(一种高级消息队列协议)的实现，这个协议主要包括一下组件

1. publisher：消息发送者
2. consumer：消息接收者
3. broker：消息中间件
4. exchange：交换机，类型主要是direct、fanout和topic
5. queue：消息队列，一个消费者只对应一个队列，一个队列对应多个消费者
6. routingkey：路由键，将queue和exchange做绑定，由consumer完成。

工作原理：首先在服务器上部署rabbitmq中间件并启动，创建消费者，通过消费者配置exchange和queue已经路由键，发送者只需知道交换机和路由键便可以发送消息

二、如何保证消息的可靠传输

由于消息传递这种方式涉及到发布者、中间件、消费者这三个组件，我们要对发布者以及消费者分别进行一些设定，来保证消息的可靠传输

(一) publisher

1. 失败回调

当发送者发出去的消息不能通过路由键找到队列时，我们可以用失败回调来将该消息返回给消费者，使用如下

• 在发送消息时设置mandatory为true
• 生产者可以通过调用channel.addReturnListener来添加ReturnListener监听器获取没有被路由到队列中的消息

2. 发布者确认

由于发布者无法得知消息是否被正确传输到exchange上，所以rabbitmq提供了两种方式来告诉用户消息是否被exchange接收，分别是事务和发布者确认。
发布者确认：当消息发送到exchange时，rabbitmq会发送ack给发布者告诉消息被接收
使用方式：一般采用设置异步监听的方式来处理，通过设置confirmListener监听器来异步处理没被exchange接收到的消息，我们可以设定是将这些消息暂时存起来定时处理或者是立刻重发

3. 事务

channel.txSelect() 将当前的信道设置成事务模式
channel.txCommit() 提交事务
channel.txRollback() 回滚事务
当消息被成功发送到exchange上事务才能提交成功，否则便可在捕获异常之后进行事务回滚，与此同时可以进行消息重发 因为事务会榨干RabbitMQ的性能，所以一般使用发布者确认代替事务

总结

我们一般通过失败回调和发布者确认的方式来保证发布者消息传递的可靠性

(二) consumer

在rabbitmq中，consumer默认是autoack(当queue将消息发出后就默认消息被正确接收)，但在实际业务中我们需要手动设置ack来保证消费者能正确接收到消息(springboot在配置文件中设置即可)

@RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "${log.info.queue}", durable = "true"),
                    exchange = @Exchange(value = "${log.exchange}", type = ExchangeTypes.TOPIC),
                    key = "${log.info.binding-key}"
            )
    )
    public void infoLog(Message message, Channel channel) throws Exception {
        String msg = new String(message.getBody());
        System.out.println(String.format("infoLogQueue 收到的消息为: {%s}", msg));
        try {
            // 这里写各种业务逻辑
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            //若未能正确接收可以发送nack来表示，若第三个参数未true表示消息重新入队，否则直接丢弃
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

若queue一直未收到ack则会一直保留消息直到这个consumer断开连接后才把该消息发送给别的consumer

publisher:发送消息时将未确认的消息暂时保存起来

@Component
public class MessageSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public final Map<String, String> unAckMsgQueue = new ConcurrentHashMap<>();

    public void convertAndSend(String exchange, String routingKey, String message) {
        String msgId = UUID.randomUUID().toString();
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(msgId);
        rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
        unAckMsgQueue.put(msgId, message);
    }

    public String dequeueUnAckMsg(String msgId) {
        return unAckMsgQueue.remove(msgId);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

confirmListener：发送成功则移除map中的消息，若失败可以重新发送或者设定自己的业务逻辑

@Component
public class ConfirmCallback implements RabbitTemplate.ConfirmCallback {

    @Autowired
    private MessageSender messageSender;

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String msgId = correlationData.getId();
        String msg = messageSender.dequeueUnAckMsg(msgId);
        if (ack) {
            System.out.println(String.format("消息 {%s} 成功发送给mq", msg));
        } else {
            System.out.println(String.format("消息 {%s} 发送mq失败", msg));
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17