从这一节开始我们进入rabbitMQ的实战环节,项目环境是spring-boot 加maven。首先让我们创建一个spring-boot项目,然后引入web依赖和 rabbitMQ的依赖

 <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>
 <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
  </dependency>

application.properties

spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.activemq.password=guest

环境搭建好之后我们就可以开始进行实战操作了。

简单消息队列

springboot会默认为你创建一个direct exchange类型交换机,其名称为""空字符串,其路由键和绑定键都是队列名称,未指定交换机的队列都会绑定到这个交换机上去。我们就以这个最简单的消息队列开始来学习如何在项目中使用rabbitMQ

我们先注册两个队列,一个用于传递String类型消息,一个传递Object类型的数据。项目启动后springboot会为你在 rabbitMQ 中创建两个队列,启动项目后打开 rabbitMQ 的 web 管理界面(以下简称管理界面)会在 Queues 中看到这两个队列的相关信息。

@Component
public class QueueConfig {
    @Bean
    public Queue getSimpleQueue() {
        return new Queue("simple-queue");
    }

    @Bean
    public Queue getObjSimpleQueue() {
        return new Queue("obj-simple-queue");
    }
}

创建两个定时任务,向 rabbitMQ 投递消息,注意这里需要在启动类上加 @EnableScheduling 注解以启动定时任务,而 Message 是我创建的实体类:

@Component
public class ScheduleHandler {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Scheduled(fixedRate = 6000)
    private void simpleQueueSchedule() {
        System.out.println("<<<<<<<<<<");

        rabbitTemplate.convertAndSend("simple-queue","ni----hao");
    }

    @Scheduled(fixedRate = 6000)
    private void objSimpleQueueSchedule() {
        System.out.println("<<<<<<<<<<");
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you ");
        rabbitTemplate.convertAndSend("obj-simple-queue",message);
    }

}

消费者消费消息:

@Component
public class QueueMessageHandler {

    @RabbitListener(queues = { "simple-queue"})
    public void getSimpleQueueMessage(String msg){
        System.out.println(msg);
    }

    @RabbitListener(queues = { "obj-simple-queue"})
    public void getObjSimpleQueueMessage(Message msg){
        System.out.println(msg);
    }

}

rabbitTemplate.convertAndSend()方法是将数据序列化并写入队列中,而其使用的序列化协议自然是java序列化协议(使用 ObjectInputStreamObjectOutputStream 读写),因此你如果调用这个方法则其实体类需要实现Serializable接口,而如果跨虚拟机还需要注意 serialVersionUID。如果跨平台了,那么最好使用其他序列化的方式,序列化反序列化配置在后文关于监听器容器的章节介绍。

推模式和拉模式

对消费端而言使用@RabbitListener监听器获取MQ消息的方式称为推模式,我们还可以使用拉模式,当我们需要一条消息的时候才从队列中拉一条消息出来,使用的方法为 rabbitTemplate.receiveAndConvert(),如:

  Message o = ((Message) rabbitTemplate.receiveAndConvert("obj-simple-queue"));

direct exchange 直连交换机

直连交换机,需要注册一个 DirectExchange , Queue , BindingBingding 负责将 DirectExchangeQueue 绑定并指定 routingKey 生产者生产消息的时候也需要指定 routingKey。下面看示例:

//  生产端配置
    @Bean("directQueueFirst")
    public Queue directQueueFirst() {
        return new Queue("first-direct-queue");
    }

    @Bean("directQueueSecond")
    public Queue directQueueSecond() {
        return QueueBuilder.durable("second-direct-queue").build();
    }
    @Bean("directExchange")
    public DirectExchange directExchange() {
        return new DirectExchange("direct-exchange");
    }

    @Bean
    public Binding bingQueueFirstToDirect(@Qualifier("directQueueFirst") Queue queue,         @Qualifier("directExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("first-key");
    }

    @Bean
    public Binding bingQueueSecondToDirect(@Qualifier("directQueueSecond") Queue queue, @Qualifier("directExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("second-key");
    }

//  生产者发送消息
@Component
public class ScheduleHandler {

    @Scheduled(fixedRate = 6000)
    private void directMessageScheduleFirst() {
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you for direct first");
        rabbitTemplate.convertAndSend("direct-exchange","first-key",message);
    }

    @Scheduled(fixedRate = 6000)
    private void directMessageScheduleSecond() {
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you for direct second");
        rabbitTemplate.convertAndSend("topic-exchange","second-key",message);
    }
}
@Component
public class QueueMessageHandler {
//  消费端
    @RabbitListener(queues = { "first-direct-queue"})
    public void firstDirectMessageQueue(Message msg){
        System.out.println(msg);
    }

    @RabbitListener(queues = { "second-direct-queue"})
    public void secondDirectMessageQueue(Message msg){
        System.out.println(msg);
    }
}

值得注意的是,springboot为了使我们的代码可读性更好,还非常贴心的提供 Exchange,Binding,QueueBuilder(建造者),因此你可以使用它们对应建造者,也可以使用直接 new 的方式进行创建。另外创建的这些 exchange queue 都能在管理界面上看到,如图 2 ,图 3 :

图 2:队列信息

​ 图 2:队列信息

图 3:交换机信息

​ 图 3:交换机信息

fanout exchange 扇型交换机

使用上和 direct exchange 大同小异,只不过不需要指定路由键,而且所有和它绑定的队列都会收到消息,直接上代码:

// 生产者配置
    @Bean("fanoutQueueFirst")
    public Queue fanoutQueueFirst() {
        return new Queue("first-fanout-queue");
    }

    @Bean("fanoutQueueSecond")
    public Queue fanoutQueueSecond() {
        return new Queue("second-fanout-queue");
    }

    @Bean("fanoutExchange")
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanout-exchange");
    }

    @Bean
    public Binding bingQueueFirstToExchange(@Qualifier("fanoutQueueFirst") Queue queue, @Qualifier("fanoutExchange") FanoutExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange);
    }

    @Bean
    public Binding bingQueueSecondToExchange(@Qualifier("fanoutQueueSecond") Queue queue, @Qualifier("fanoutExchange") FanoutExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange);
    }
@Component
public class ScheduleHandler {
//  生产者发消息,注意这里虽然填了routingKey 但是是无效的
    @Scheduled(fixedRate = 6000)
    private void directMessageScheduleFirst() {
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you for direct first");
        rabbitTemplate.convertAndSend("direct-exchange","first-key",message);
    }
}
//  消费者,两个队列都能收到同一份消息
@Component
public class QueueMessageHandler {
    @RabbitListener(queues = { "first-fanout-queue"})
    public void firstFanoutQueue(Message msg){
        System.out.println(msg);
    }

    @RabbitListener(queues = { "second-fanout-queue"})
    public void secondFanoutQueue(Message msg){
        System.out.println(msg);
    }
}

主题交换机 Topic

前文介绍了主题交换机的路由方式,注意我代码中的路由键设置,这里我设置两个bingding-key 分别是 com.muggle.firstcom.# 我用 routing-keycom.muggle.test 发消息这两个队列都能接收到

    @Bean("topicQueueFirst")
    public Queue topicQueueFirst() {
        return new Queue("first-topic-queue");
    }

    @Bean("topicQueueSecond")
    public Queue topicQueueSecond() {
        return new Queue("second-topic-queue");
    }

    @Bean
    public Binding bindTopicFirst(@Qualifier("topicQueueFirst") Queue queue, @Qualifier("topicExchange") TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("com.muggle.first");
    }

    @Bean
    public Binding bindTopicSecond(@Qualifier("topicQueueFirst") Queue queue, @Qualifier("topicExchange") TopicExchange exchange) {
        return BindingBuilder.bind(topicQueueFirst()).to(topicExchange()).with("com.#");
    }

@Component
public class ScheduleHandler {
    @Scheduled(fixedRate = 6000)
    private void topicMessage() {
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you for topic test");
        rabbitTemplate. convertAndSend("topic-exchange","com.muggle.test",message);
    }

}

@Component
public class QueueMessageHandler {
    @RabbitListener(queues = { "first-topic-queue"})
    public void firstTopicMessageQueue(Message msg){
        System.out.println(msg);
    }

    @RabbitListener(queues = { "second-topic-queue"})
    public void secondTopicMessageQueue(Message msg){
        System.out.println(msg);
    }
}

现在我们学习了 rabbitMQ 的各类交换机的用法,这些只是 rabbitMQ 的基础特性,下文我们将介绍一些 rabbitMQ 的更复杂的使用方法。


作者:muggle 点我关注作者

出处:https://muggle-book.gitee.io/

版权:本文版权归作者所有

转载:欢迎转载,但未经作者同意,必须保留此段声明;必须在文章中给出原文连接;否则必究法律责任

results matching ""

    No results matching ""