本文导航

RabbitMQ和Spring AMQP学习一

下载和安装

详细步骤详见官方文档

基本概念

背景知识

最初的消息模型只有三个要素，生产者、消息队列、消费者。随着消息队列的演进，RabbitMQ变为现在的消息模型。主要包含的要素为，生产者、routingKey、Exchange、Binding、Queue、消费者.

官方教程里的话也是充分说明了完整的消息模型

Producer

生产者：顾名思义，其作用就是生产消息。在RabbitMQ消息模型中，生产者生产的消息，会发送到Exchange。不关心谁消费。生产者眼里只有Exchange.

Routing Key

生成者在将消息发送给Exchange时，可能携带着routingKey,Exchange会根据消息中的routingKey和另一边的bindingKey根据当前Exchange的类型进行匹配。如果满足规则，则会将消息转发给bindingKey对应的Queue

Exchange

交换器.一边连着生产者，一边连着Queue.交换器的职责就是接收消息，并根据消息中携带的routingKey找到对应bindingKey的消息队列。并把消息转发出去。

Exchange的type分为：

• 广播(fanout)即将收到的消息转发给所有连接在此Exchange上的Queue。
• 完全匹配(direct)即当routingKey和bindingKey完全一样时才转发给对应的Queue
• 模式匹配(topic)

Binding

Exchange和Queue之间的绑定关系。在RabbitMQ中称为Binding，具体的值称为bindingKey

Queue

消息队列，真正存储消息的地方。如果此消息队列上有消费者监听，会进行消息的消费

控制台的基本使用

控制台的默认端口为15672

设置virtual hosts

默认是/，如果想要控制不同的角色的连接。则可以在控制台添加配置即可

一般如果virtual-host配置和对应的角色对应不上，是连接不上的，一般会Connection rest。这时要检查下当前连接RabbitMQ的用户所对应的virtual-host是什么

通过控制台发送消息(发送时报No method found for class [B==解决方法)

在实际编码过程中，编写完毕之后，测试消费者是否可以正常接收到消息时，可以在对应的Exchange或者Queue下手动编写消息体进行发送

在对应的payload中编写消息体。这里要注意，在控制台发送消息时，一般要指定Properties,如果消费者用String接收，则Properties为content_type=text/plain.如果不设置会怎么样，一般都会报Caused by: org.springframework.amqp.AmqpException: No method found for class [B这种错误

Spring AMQP基本使用

配置Exchange、Queue、Binding

@Configuration
public class DirectExchangeConfig {

    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("direct");
    }

    @Bean
    public Queue firstDirectQueue() {
        return new Queue("firstDirect");
    }

    @Bean
    public Queue secondDirectQueue() {
        return new Queue("secondDirect");
    }

    @Bean
    public Binding firstDirectBinding() {
        return BindingBuilder.bind(firstDirectQueue()).to(directExchange()).with("first");
    }

    @Bean
    public Binding secondDirectBinding() {
        return BindingBuilder.bind(secondDirectQueue()).to(directExchange()).with("second");
    }
}

当前Exchange的类型为Direct,即完全匹配。通过BindingBuilder.bind(队列).to(exchange).with(routingKey)完成队列和Exchange之间的绑定

消费者监听消息队列，以便随时消费任务

@Component
@Slf4j
public class DirectConsumer {

    @RabbitListener(queues = "firstDirect")
    @RabbitHandler
    public void receiveFirstDirectQueueMessage(String firstQueueMessage) {
        log.info("This is firstDirectQueue received message: " + firstQueueMessage);
    }

    @RabbitListener(queues = "secondDirect")
    @RabbitHandler
    public void  receiveSecondDirectQueueMessage(String secondQueueMessage) {
        log.info("This is secondDirectQueue received message: " + secondQueueMessage);
    }
}

生产者向Exchange发送消息

携带不同的routingKey，会被Direct Exchange转发的对应的Queue上

@Component
public class DirectProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private DirectExchange directExchange;

    public void sendDirectMessage() {
        rabbitTemplate.convertAndSend(directExchange.getName(), "first", "this is send first message");
        rabbitTemplate.convertAndSend(directExchange.getName(), "second", "this is send second message");
    }
}

小技巧

在配置文件中，将rabbitmq的日志级别设置为debug。可以看到消息发送过程中打印出的日志

logging:
  level:
    org.springframework.amqp: debug

下一篇：Rabbit MQ消息模型详解