SpringBoot集成rabbitMq 具体的编程操作 Direct Exchange

朱雀 2022-10-07 14:49 209阅读 0赞

**[SpringBoot集成rabbitMq入门（目录）][SpringBoot_rabbitMq]**  
首先，所有的操作都要引入jar依赖

<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.yml** 配置文件

server:
      port: 8021
    spring:
      application:
        name: rabbitmq-provider				#给项目来个名字
      rabbitmq:								#配置rabbitMq 服务器
        host: 127.0.0.1						#ip地址
        port: 5672							#端口号
        username: admin
        password: 123456					#用户名密码，这是我自己创建的
        virtual-host: /						#虚拟host 可以不设置,使用server默认host，也可以使用自己创建的虚拟主机名称

## 交换机与消息队列的绑定 ##

package com.hnbd.dome.config;
    
    import com.hnbd.dome.constant.RabbitConstants;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.annotation.EnableRabbit;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * RabbitMqConfig
     *  连接 消息队列的 基本配置  这里使用的消息队列是 RabbitMq
     * @创建人 江枫沐雪
     * @创建时间 2021/6/16 9:37
     */
    @Configuration       //用于定义配置类  可代替xml配置文件
    @EnableRabbit        //监听消息队列的内容
    public class RabbitMqConfig {
        
    
        /**
         * 直连型交换机  direct
         * durable：是否声明一个持久的交换机（服务器重启后 依然存活）
         * autoDelete：是否自动删除。
         * @return
         */
        @Bean
        public DirectExchange directExchange() {
        
            return new DirectExchange("direct_test", true, false);
        }
    
        /**
         * 队列
         * @return
         */
        @Bean
        public Queue position() {
        
            // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
            // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
            // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
            //   return new Queue("TestDirectQueue",true,true,false);
            
            //一般设置一下队列的持久化就好,其余两个就是默认false
            return new Queue("QUEUE_TEST", true);
        }
    
        /**
         * 绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
         * @param directExchange            交换机
         * @param position                  队列
         * @return
         */
        @Bean
        public Binding binding(DirectExchange directExchange, Queue position) {
        
            return BindingBuilder.bind(position).to(directExchange).with("Binding_test");
        }
    }

## 生产者 发送消息 ##

传递两种不同的参数

package com.hnbd.dome.controller;
    
    import com.hnbd.dome.constant.RabbitConstants;
    import com.hnbd.dome.rabbit.RabbitSender;
    import lombok.SneakyThrows;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.UUID;
    
    /**
     * SendMessageController    将消息发送到消息队列中    生产者
     *
     * @创建人 江枫沐雪
     * @创建时间 2021/6/18 9:09
     */
    @Slf4j              //日志
    @RestController
    public class SendMessageController {
        
     
        @Autowired
        private RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法
    
        /**
         * 普通的消息发送
         * @return
         */
        @SneakyThrows
        @GetMapping("/sendDirectMessage")
        public String sendDirectMessage() {
        
    
            String messageId = String.valueOf(UUID.randomUUID());   //随机生成一的id，利用了工具类，可以注释掉
            String messageData = "test message, hello!";
            Map<String,Object> map=new HashMap<>();
            map.put("messageId",messageId);
            map.put("messageData",messageData);
    
            //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
            rabbitTemplate.convertAndSend("direct_test","Binding_test", map);
            return "ok";
        }
    
        /**
         * 消息测试
         * @return
         */
        @GetMapping("/test")
        public String testConfirm() {
        
            rabbitTemplate.convertAndSend("direct_test","Binding_test", "你好！世界");
            return "ok";
        }
    }

测试  
![在这里插入图片描述][20210621115636497.png]  
因为我们目前还没弄消费者 rabbitmq-consumer，消息没有被消费的，我们去rabbitMq管理页面看看，是否推送成功：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h0MzE0MTU5_size_16_color_FFFFFF_t_70]

## 消费者 ##

@Component
    @RabbitListener(queues = "QUEUE_TEST")    //监听的队列名称 TestDirectQueue
    public class DirectReceiver {
        
     
        @RabbitHandler
        public void process(StringtestMessage) {
        
            System.out.println("DirectReceiver消费者收到消息  : " + testMessage);
        }
     
    }

项目运行起来，可以看到把之前推送的那条消息消费下来了：  
![在这里插入图片描述][20210621141248714.png]

[SpringBoot_rabbitMq]: https://blog.csdn.net/xt314159/article/details/117999149
[20210621115636497.png]: /images/20221005/a7234150ab134e15b55a6a1ab15672f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h0MzE0MTU5_size_16_color_FFFFFF_t_70]: /images/20221005/00eb12be4607464c882f894ac76b7893.png
[20210621141248714.png]: /images/20221005/520217fb93d947808837e0ccfd30fbd9.png