RabbitMQ从入门到实战：消息队列核心用法详解

本文基于作者实际开发经验，已在项目中使用 RabbitMQ 实现异步消息处理，涵盖交换机配置、队列监听及消费者实践。

最近处理访客记录所以，来学习下RabbitMQ。之前同事已经写好了，这里只需要进行消费，后续会逐渐完善。

0.介绍#

0.1 交换机（Exchanges）#

RabbitMQ中生产者发送的消息都是发送到交换机，再由交换机推入队列。所以生产者不知道队列去了哪里，就靠Exchange来控制，交换机总共有以下几种类型。

0.1.1 广播模式（fanout）#

扇出所有消息进入队列，类似广播。

0.1.2 直接交换（direct）#

绑定相关的routerKey分发到不同的队列，简单说就是direct交换机接收了消息后，根据关键词分发队列。

0.1.3 主题模式（topic）#

direct路由比较单一，所以提升了routerKey的能力，在关键词标记下加上了通配符。

*（星号）可以代替一个单词
#（井号）可以替代零个或多个单词

1.公共配置类#

1
spring:
2
    rabbitmq:
3
        host: 127.0.0.1
4
        port: 5672
5
        username: guest
6
        password: guest

1
/**
2
 * RabbitMQ公共配置类
3
 */
4
public class RabbitMQConfig {
5
    /** RabbitMQ的队列主题名称 */
6
    public static final String RABBITMQ_TOPIC = "rabbitmqTopic";
7
    /** RabbitMQ的DIRECT交换机名称 */
8
    public static final String RABBITMQ_DIRECT_EXCHANGE = "rabbitmqDirectExchange";
9
    /** RabbitMQ的Direct交换机和队列绑定的匹配键 */
10
    public static final String RABBITMQ_DIRECT_ROUTING = "rabbitmqDirectRouting";
11
}

2.消费消息的两种方式#

把记录塞进队列里的时候，只是完成了第一步，那你肯定要对他进行消费。分为两种推模式和拉模式：推模式就是生产者发布消息时，主动推送给消费者；拉模式则是消费者发送请求后才会发送。

3.监听队列的两种方式#

一种是@RabbitListener注解的方式，一种是实现SpringBoot的ChannelAwareMessageListener接口的方式。

3.1 @RabbitListener#

如果demoData想不转换成String直接推，得在这个数据流实现序列化。

1
innerRabbitTemplate.convertAndSend(InnerMQConfig.TOPIC_EXCHANGE, msgKey, JSONObject.toJSONString(demoData));

1
@Component
2
public class DemoRabbitMQListener {
3
      @RabbitListener(queues = "demo_queue")
4
      @RabbitHandler
5
      public void demoQueue(Message message){
6
          System.out.println("message:"+message.getBody());
7
      }
8
}

3.2 实现ChannelAwareMessageListener接口#

听前辈说直接实现这个接口，就不用管底层是谁的消息队列了，因为是基于SpringBoot。这个实现起来有点麻烦，我总结了以下顺序：

3.2.1 创建连接工厂（ConnectionFactory）#

publisherConfirms：消息发送到exchange，返回成功或者失败。 publishReturns：消息从exchange到queue，发送成功或者失败。

1
    @Bean(name = "DemoConnectionFactory")
2
    @Primary
3
    public ConnectionFactory connectionFactory() {
4
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
5
        connectionFactory.setHost(host);
6
        connectionFactory.setPort(port);
7
        connectionFactory.setUsername(username);
8
        connectionFactory.setPassword(password);
9
        connectionFactory.setVirtualHost(virtualHost);
10
        connectionFactory.setPublisherConfirms(true);
11
        connectionFactory.setPublisherReturns(true);
12
        return connectionFactory;
13
    }

3.2.2 初始化组件（RabbitAdmin）#

1
    @Bean(name = "DemoRabbitAdmin")
2
    @Primary
3
    public RabbitAdmin rabbitAdmin(@Qualifier("DemoConnectionFactory") ConnectionFactory connectionFactory) {
4
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
5
        rabbitAdmin.setAutoStartup(true);
6
        return rabbitAdmin;
7
    }

3.2.3 创建交换器（Exchange）#

durable：是否持久化，RabbitMQ关闭后，没有持久化的Exchange将被清除 autoDelete：是否自动删除，如果没有与之绑定的Queue，直接删除 internal：是否内置的，如果为true，只能通过Exchange到Exchange

1
    @Bean(DEMO_EXCHANGE)
2
    public TopicExchange exchange() {
3
        return new TopicExchange(DEMO_EXCHANGE, true, false);
4
    }

3.2.4 创建队列（Queue）#

创建队列主要掌握这几个参数：

name：队列名称
durable：队列是否持久化
exclusive：是否排他的队列
autoDelete：是否自动删除
arguments：队列中的消息什么时候会自动被删除

1
    @Bean(QUEUE_NAME)
2
    public Queue QUEUE_DEMO() {
3
        return new Queue(QUEUE_NAME, true, false, false);
4
    }

3.2.5 绑定队列到交换机（Binding）#

1
    @Bean
2
    public Binding BINGING_EXCHANGE_QUEUE(@Qualifier(QUEUE_NAME) Queue queue,
3
                                           @Qualifier(DEMO_EXCHANGE) Exchange exchange) {
4
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY).noargs();
5
    }

3.2.6 创建监听容器（SimpleMessageListenerContainer）#

1
    @Bean
2
    public SimpleMessageListenerContainer simpleMessageListenerContainer(
3
            @Qualifier("DemoConnectionFactory") ConnectionFactory connectionFactory,
4
            DemoRabbitMQListener demoRabbitMQListener,
5
            @Qualifier(QUEUE_NAME) Queue queue
6
    ) throws AmqpException {
7
        SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
8
        listenerContainer.setConcurrentConsumers(listenerSize);
9
        listenerContainer.setQueues(queue);
10
        listenerContainer.setExposeListenerChannel(true);
11
        listenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
12
        listenerContainer.setMessageListener(demoRabbitMQListener);
13
        return listenerContainer;
14
    }

3.2.7 创建操作类（RabbitTemplate）#

setConfirmCallback的消息回调是在生产者端要把参数丢进去的。

1
    @Bean(name = "DemoRabbitTemplate")
2
    @Primary
3
    public RabbitTemplate rabbitTemplate(@Qualifier("DemoConnectionFactory") ConnectionFactory connectionFactory) {
4
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
5
        rabbitTemplate.setMandatory(true);
6
        rabbitTemplate.setConnectionFactory(connectionFactory);
7
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
8
            @Override
9
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
10
                if (ack) {
11
                    log.info("消息发送到交换机成功, 消息 = {}", correlationData.getId());
12
                } else {
13
                    log.error("消息发送到交换机失败! 消息: {}; 错误原因: cause: {}", correlationData.getId(), cause);
14
                }
15
            }
16
        });
17
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
18
            @Override
19
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
20
                String messageId = message.getMessageProperties().getMessageId();
21
                String result = null;
22
                try {
23
                    result = new String(message.getBody(), "UTF-8");
24
                } catch (Exception e) {
25
                    log.error("消息发送失败", e);
26
                }
27
                log.error("消息发送失败, 消息ID = {}; 消息内容 = {}", messageId, result);
28
            }
29
        });
30
        return rabbitTemplate;
31
    }

3.2.8 监听消费（RabbitMQListener）#

这个类要注意用@Service或者@Component注解让他交给IOC

1
@Service
2
@Slf4j
3
public class DemoRabbitMQListener implements ChannelAwareMessageListener {
4

5
    @Override
6
    public void onMessage(Message message, Channel channel) throws Exception {
7
        log.info("message:{}", message.getBody());
8
        //todo: 接下来就是各自的业务逻辑，就是消费环节
9
    }
10
}