SpringBoot整合RabbitMQ（创建队列、一次性消费、发送消息、路由绑定，监听消费）

文章目录

- 主要步骤
- 环境
- 实际操作
- - 创建SpringBoot工程
  - 配置RabbitMQ的基本参数
  - 操作RabbitMQ
  - RabbitMQ监听消费
- 场景模拟
- - 具体实现

主要步骤

创建SpringBoot工程
配置RabbitMQ的基本参数
host 、virtual host、username、password
通过RabbitTemplate创建RabbitAdmin，通过RabbitAdmin操作RabbitMQ
RabbitAdmin可以创建队列、交换机、发送数据、主动消费数据
同时SpringBoot提供了Listener监听器来实现实时消费数据

环境

开发工具：IDEA 2020.1
操作系统：Centos7
RabbitMQ版本：Docker中rabbitmq:management
Docker版本：20.10.9

实际操作创建SpringBoot工程

使用命令创建SpringBoot工程

mvn org.apache.maven.plugins:maven-archetype-plugin:3.1.2:generate -DarchetypeArtifactId="maven-archetype-webapp" -DarchetypeGroupId="org.apache.maven.archetypes" -DarchetypeVersion="1.4"

该命令创建的是传统maven工程，需要手动改造成SpringBoot工程
删除WebApp目录及下属内容，改造为下列文件结构

添加pom文件中的依赖
完全改造后内容如下，比对完成即可




  4.0.0

  com.dean
  SpringBoot_RabbitMQ
  1.0-SNAPSHOT
  
    org.springframework.boot
    spring-boot-starter-parent
    2.2.5.RELEASE
  



  
    1.8
  

  
    
      org.springframework.boot
      spring-boot-starter-web
    
    
      org.springframework.boot
      spring-boot-starter-amqp
    
    
      org.springframework.boot
      spring-boot-starter-test
    
    
      junit
      junit
      test
    
  

  
    SpringBoot_RabbitMQ
    
      
        
          maven-clean-plugin
          3.1.0
        
        
        
          maven-resources-plugin
          3.0.2
        
        
          maven-compiler-plugin
          3.8.0
        
        
          maven-surefire-plugin
          2.22.1
        
        
          maven-war-plugin
          3.2.2
        
        
          maven-install-plugin
          2.5.2
        
        
          maven-deploy-plugin
          2.8.2

主要是、、节点进行了更改

编写启动类

配置RabbitMQ的基本参数

resources目录下创建application.yml配置文件
编写参数

操作RabbitMQ

创建一个SpringBoot的测试用例
上面的操作中，我们已经在pom文件中添加了SpringBoot Test的依赖了
现在我们需要src目录下创建一个test目录，在test目录下写一个RabbitTest类进行测试
创建一个队列

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitApplication.class)
public class RabbitTest {
    @Resource
    private RabbitTemplate rabbitTemplate;

    //声明一个队列
    @Test
    public void createQueue() {
        //创建RabbitAdmin 用于操作队列
        RabbitAdmin admin = new RabbitAdmin(rabbitTemplate);
        Queue springQueue = new Queue("SpringQueue");
        admin.declareQueue(springQueue);
    }
}

执行后，可以在RabbitMQ的web界面看到新生成的队列

向队列发送消息

 @Test
    public void sendMsg() {
        rabbitTemplate.convertAndSend("SpringQueue","Hello Spring Rabbit");
    }
    }

消费（一次性消费）
这个场景就是例如我们上线某聊天软件后，需要得知有多少未读信息，不去拿消息本身

@Test
    public void consumeMsg() {
        rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        //队列的属性 Properties继承HashTable
        Properties properties = rabbitAdmin.getQueueProperties("SpringQueue");
        //由于不知道properties里面都有什么，所以输出看一下
        
        Enumeration names = properties.propertyNames();
        while (names.hasMoreElements()) {
            Object e = names.nextElement();
            System.out.println(e);
        }
        //获取消息总数
        String msg = properties.get("QUEUE_MESSAGE_COUNT").toString();
        System.out.println(msg );
        //清空队列 （也就是说本次是一次性消费）
        rabbitAdmin.purgeQueue("SpringQueue");
    }

消息已被清空

路由绑定

@Test
    public void routeBind() {
        rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        //声明队列
        //系统消息接收队列，userId在实际生产中为实际用户id
        Queue sysQueue = new Queue("sys.msg.userId");
        //一对一消息接收队列，sendUserId为实际向用户发送消息的用户id
        Queue userQueue=new Queue("user.msg.sendUserId");
        //声明交换机 Direct
        DirectExchange directEx = new DirectExchange("notice.direct");
        //声明绑定
        Binding sysBind = BindingBuilder.bind(sysQueue).to(directEx).with("sys");
        Binding userBind = BindingBuilder.bind(userQueue).to(directEx).with("user");

        //创建队列
        rabbitAdmin.declareQueue(sysQueue);
        rabbitAdmin.declareQueue(userQueue);
        //创建交换机
        rabbitAdmin.declareExchange(directEx);
        //创建绑定关系
        rabbitAdmin.declareBinding(sysBind);
        rabbitAdmin.declareBinding(userBind);
    }

成功创建的队列

成功创建的交换机及绑定关系

通过路由向队列发送消息

@Test
    public void sendMsg2Route() {
        rabbitTemplate.convertAndSend("notice.direct","sys", "系统通知");
        rabbitTemplate.convertAndSend("notice.direct","sys", "系统通知");
        rabbitTemplate.convertAndSend("notice.direct","user", "用户通知");
    }

向系统队列发两条消息，向用户队列发一条消息

RabbitMQ监听消费

首先我们需要一个监听器和一个监听配置
监听器SysNoticeListener

@Component
public class SysNoticeListener implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        System.out.println(Thread.currentThread().getName()+": 监听到消息");
        Thread.sleep(1000);
    }
}

重点在于实现了ChannelAwareMessageListener类，重写了其方法OnMessage

监听配置类 RabbitConfig

@Configuration
public class RabbitConfig {
    //注入自定义的监听器
    @Resource
    private SysNoticeListener sysNoticeListener;

    @Bean
    public SimpleMessageListenerContainer listenerContainer(ConnectionFactory connectionFactory) {
        //创建监听器工厂
        SimpleRabbitListenerContainerFactory factory=new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);

        //ack机制  自动消费 或 手动消费
        //自动消费：无论消费是否成功都进行确认
        //手动消费：由程序员来判断消费是否成功
        //设置为无ACK机制
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);

        //设定消费者的线程数
        factory.setMaxConcurrentConsumers(5); //设置最大其实也不生效
        factory.setConcurrentConsumers(3);  //3个消费者

        //使用工厂生成监听器容器
        SimpleMessageListenerContainer container = factory.createListenerContainer();
        //绑定监听队列  监听器监听哪个队列
        container.setQueueNames("sys.msg.userId");
        //绑定监听器  用哪个监听器来监听
        container.setMessageListener(sysNoticeListener);
        return container;
    }
}

当队列中有消息时，会进行消费，程序运行后like this

场景模拟

我们虚拟一个场景，用RabbitMQ实现一下
故事是这样的，当用户上线时,获得系统未读消息数量，上线期间，可以实时获取消息，下线后不再实时获取

也就是说，当用户上线时，返回当前队列中消息总数，上线期间，实时监听消息

具体实现

编写RabbitConfig

@Configuration
public class RabbitConfig {
    //注入自定义的监听器
    @Resource
    private SysNoticeListener sysNoticeListener;

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Bean("sysListenerContainer")
    public SimpleMessageListenerContainer listenerContainer(ConnectionFactory connectionFactory) {
        //创建监听器工厂
        SimpleRabbitListenerContainerFactory factory=new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);

        //ack机制  自动消费 或 手动消费
        //自动消费：无论消费是否成功都进行确认
        //手动消费：由程序员来判断消费是否成功
        //设置为无ACK机制
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);


        //设定消费者的线程数
        factory.setMaxConcurrentConsumers(5); //设置最大其实也不生效
        factory.setConcurrentConsumers(3);  //3个消费者

        //使用工厂生 成监听器容器
        SimpleMessageListenerContainer container = factory.createListenerContainer();
        //绑定监听队列  监听器监听哪个队列
        //container.setQueueNames("sys.msg.userId");
        //绑定监听器  用哪个监听器来监听
        container.setMessageListener(sysNoticeListener);
        return container;
    }

    
    @Bean("rabbitAdmin")
    public RabbitAdmin rabbitAdmin() {
        RabbitAdmin admin = new RabbitAdmin(rabbitTemplate);
        List queues = new ArrayList<>();
        //创建队列
        IntStream.range(0,2).forEach(i->{
            queues.add(new Queue("sys.msg."+i));
        });

        //创建交换机
        //因是系统通知，所以创建Fanout交换机
        FanoutExchange exchange = new FanoutExchange("sys.notice.fanout");

        //交换机和队列绑定
        List bindings = new ArrayList<>();
        for (Queue q : queues) {
            bindings.add(BindingBuilder.bind(q).to(exchange));
        }

        admin.declareExchange(exchange);
        IntStream.range(0,2).forEach(i->{
            admin.declareQueue(queues.get(i));
            admin.declareBinding(bindings.get(i));
        });
        return admin;
    }

}

rabbitAdmin方法中可以根据实际情况，增大循环数，为每一个用户创建一个接收系统通知的队列

编写Controller

@RestController
@RequestMapping("user")
public class ConsumeController {
    //注入容器
    @Resource(name = "sysListenerContainer")
    private SimpleMessageListenerContainer container;
    //注入监听器
    @Resource
    private SysNoticeListener listener;
    //注入RabbitAdmin
    @Resource(name = "rabbitAdmin")
    private RabbitAdmin rabbitAdmin;

    
    @GetMapping("login/{userId}")
    public String login(@PathVariable String userId) {
        

        //根据UserId拼接自己的队列名称
        String queueName = "sys.msg." + userId;
        Properties queueProperties = rabbitAdmin.getQueueProperties(queueName);
        //通过属性值获取消息总量
        int count = Integer.parseInt(queueProperties.get("QUEUE_MESSAGE_COUNT").toString());
        System.out.println("您未上线的时间里，有" + count + "条消息在等您");
        //清空队列 一定要清空
        rabbitAdmin.purgeQueue(queueName);

        

        //监听容器绑定监听队列
        container.setQueueNames(queueName);

        return "success";

    }


    
    @GetMapping("/logout/{userId}")
    public String logout(@PathVariable("userId") String userId) {
        String queueName = "sys.msg." + userId;
        container.removeQueueNames(queueName);
        return "logout success";
    }
}

也就是在用户成功登录后才进行将监听容器和队列相绑定，在用户退出登录后，从监听容器中移除该队列

编写监听器Listener

@Component
public class SysNoticeListener implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {

       

        //根据队列名称获取用户id
        MessageProperties messageProperties = message.getMessageProperties();
        //获取队列名称
        String consumerQueue = messageProperties.getConsumerQueue();
        //获取消息体
        byte[] body = message.getBody();
        System.out.println(consumerQueue + "：收到实时消息【" + new String(body) + "]");
    }
}

监听器实际上处理的是实时消息的处理

最终效果

登录哪一个用户，哪一个用户就可以接收实时消息
补充一点关于ACK机制
如果在RabbitConfig中设置了ACK为手动模式，那么就需要在监听器中增加try-catch进行手动处理

RabbitConfig中

监听器中

消息的处理需要放在try中，如果消息消费失败，则将消息返回给队列

SpringBoot整合RabbitMQ（创建队列、一次性消费、发送消息、路由绑定，监听消费）

大数据系统相关栏目本月热门文章