【2023】RabbitMQ配合springboot使用代码实现

桃扇骨 2023-10-13 17:24 51阅读 0赞

### 使用实现 ###

*  YML配置文件

#  设置端口号
    server:
      port: 8089  
    	#rabbitQM配置
    spring:	
      rabbitmq:
    		virtual-host: /      #角色
        addresses: 192.168.204.129
        port: 5672
        username: root
        password: root

### 1、direct模式 ###

*  使用direct模式直发
    
    Direct是RabbitMQ默认的交换机模式,也是最简单的模式.即创建消息队列的时候,指定一个BindingKey.当发送者发送消息的时候,指定对应的Key.当Key和消息队列的BindingKey一致的时候,消息将会被发送到该消息队列中.
    
    1.  一对一
        
        1.  config：配置类
            
            把队列注入容器，初始化时创建：类上需要加配置类注解-`@Configuration`
            
            队列的依赖包名：`import org.springframework.amqp.core.Queue;`
            
                /**
                 *@Description//TODO把队列注入到容器中，让初始化时创建这个队列
                *@Date2022/8/9 15:07
                 *@return:org.springframework.amqp.core.Queue
                 **/
                @Bean
                public Queue paymentNotifyQueue(){
                          
                    return new Queue("notify.payment");
                }
        2.  sender：生产者
            
            发送消息：`AmqpTemplate`
            
                /**
                 *@program:RabbitMQ
                 *@description:生产者-队列发送消息
                *@author:zheng
                 *@create:2022/08/09 14:32
                 */
                @Component
                public class PaymentNotifySender {
                          
                
                    @Autowired
                    private AmqpTemplate amqpTemplate;
                
                    public void sender(String msg){
                          
                        System.err.println("notify.payment 发送消息："+msg);
                //          发送消息到队列：队列名，发送的参数
                        amqpTemplate.convertAndSend("notify.payment",msg);
                    }
        3.  recive：消费者
            
            接收消息方法参数类型需要和发送的类型一致
            
                /**
                 *@program:RabbitMQ
                 *@description:消费者-接收队列消息
                *@author:zheng
                 *@create:2022/08/09 14:43
                 *@RabbitListener监听的队列
                */
                @Component
                public class PaymentNotifyRecive {
                          
                /**
                *@Description//TODO消费者消息处理的方法
                *@param:msg从队列中接收的消息
                * RabbitHandler表示该方法为接收消息的方法
                **/
                @RabbitListener(queues = "notify.payment")
                    @RabbitHandler
                    public void receive(String msg){
                          
                        System.out.println("接收来自notify.payment的消息："+msg);
                    }
    2.  一对多：
        
        一对多的话，多个接收者之间采用的是轮询的方式进行接收，

### 2、Topic模式 ###

*  使用交换机的Topic模式
    
    topic转发信息主要是依据通配符,队列和交换机的绑定。主要是依据一种模式(通配符+字符串),而当发送消息的时候,只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中.
    
    1.  配置类：config
        
         *  使用交换机需要注入bean三个对象，一个队列对象(Queue)，一个交换机对象(TopicExchange)，一个绑定对象(Binding)
         *  一个交换机可以绑定多个队列
         *  绑定对象中的参数队列名，需要和Queue注入bean的方法名一致
        
            /**
             * @program: RabbitMQ
             * @description: 注入交换机
             * @author: zheng
             * @create: 2022/08/09 15:53
             */
            @Configuration
            public class ToipcConfig {
                    
            
            /**
             *@Description//TODO注入队列
            *@return:org.springframework.amqp.core.Queue
             **/
            @Bean
            public Queue coreQueue(){
                    
                return new Queue("api.core");
            }
            /**
             *@Description//TODO注入交换机
            *@return:org.springframework.amqp.core.TopicExchange
             **/
            @Bean
            public TopicExchange coreExchange(){
                    
                return new TopicExchange("coreExchange");
            }
            
            /**
             *@Description//TODO绑定交换机和队列加路由键
            *@param:coreQueue绑定的队列
            *@param:coreExchange绑定的交换机
            *@return:org.springframework.amqp.core.Binding返回绑定对象
            **/
            @Bean
            public Binding bindingCore(Queue coreQueue , TopicExchange coreExchange){
                    
                return BindingBuilder.bind(coreQueue).to(coreExchange).with("api.core.*");
            }
            }
        
        1.  生产者：
            
             *  `RabbitTemplate`为`AmqpTemplate`接口的实现类
             *  只有路由键满足条件的队列才会接收到
        
            /**
             *@program:RabbitMQ
             *@description:用户管理
            *@author:zheng
             *@create:2022/08/09 15:52
             */
            @Component
            public class ApiCoreSender {
                    
                @Autowired
                private RabbitTemplate rabbitTemplate;
            /**
            *用户管理方法
            *@parammsg
            */
            public void user(String msg){
                    
                    System.out.println("api.core.user send message: "+msg);
                    /*
                     *发送消息：交换机、路由键、消息内容
                     */
                    rabbitTemplate.convertAndSend("coreExchange","api.core.user",msg);
                }
            
            /**
                 *查询用户方法
            *@parammsg
            */
            public void userQuery(String msg){
                    
                    System.out.println("api.core.user.query send message: "+msg);
                    rabbitTemplate.convertAndSend("coreExchange","api.core.user.query");
                }
            }
        
        1.  消费者
        
            /**
             *@program:RabbitMQ
             *@description:交换机模式=接收者
            */
            @Component
            public class ApiCoreRecive {
                    
            /**
                 *接收用户发送到交换机的消息
            * RabbitListener：监听的队列
            * RabbitHandler：表示监听消息
            *@parammsg
            */
            @RabbitListener(queues = "api.core")
                @RabbitHandler
                public void getUser(String msg){
                    
                    System.out.println("api.core.user 接收到的消息:"+msg);
                }
            }

### 3、Fanout模式 ###

*  使用交换机的Fanout模式
    
    一个交换机绑定多个队列
    
    1.  配置类：config
        
            /**
             *@program:RabbitMQ
             *@description:fanout注入对象
            *@author:zheng
             *@create:2022/08/09 17:22
             */
            @Configuration
            public class FanoutConfig {
                    
            //队列
                @Bean
                public Queue reportPaymentQueue(){
                    
                    return new Queue("api.report.payment");
                }
                @Bean
                public Queue reportRefundQueue(){
                    
                    return new Queue("api.report.refund");
                }
            //交换机
                @Bean
                public FanoutExchange reportExchange(){
                    
                    return new FanoutExchange("reportExchange");
                }
            
            /**
                 *@Description//TODO绑定对象-Fanout广播模式
            *@param:paymentQueue
                 *@param:paymentExchange
                 *@return:org.springframework.amqp.core.Binding
                 **/
            @Bean   
                public Binding bindingReportPaymentExchange(Queue reportPaymentQueue, FanoutExchange reportExchange){
                    
                    return BindingBuilder.bind(reportPaymentQueue).to(reportExchange);
                }
                @Bean
                public Binding bindingReportRefundExchange(Queue reportRefundQueue, FanoutExchange reportExchange){
                    
                    return BindingBuilder.bind(reportRefundQueue).to(reportExchange);
                }
            }
    2.  生产者：sender
        
            /**
             *@program:RabbitMQ
             *@description:发送报表
            *@author:zheng
             *@create:2022/08/09 17:20
             */
            @Component
            public class ApiReportSender {
                    
                @Autowired
                private AmqpTemplate amqpTemplate;
            
                public void generateReports(String msg){
                    
                    System.out.println("api.generate.reports 发送消息: "+msg);
                    amqpTemplate.convertAndSend("reportExchange","",msg);
                }
            }
    3.  消费者：recive
        
            /**
             *@program:RabbitMQ
             *@description:接收报表
            *@author:zheng
             *@create:2022/08/09 17:26
             */
            @Component
            public class ApiReportRecive {
                    
                @RabbitListener(queues = "api.report.payment")
                @RabbitHandler
                public void payment(String msg){
                    
                    System.out.println("接收用户报表："+msg);
                }
            
                @RabbitListener(queues = "api.report.refund")
                @RabbitHandler
                public void refund(String msg){
                    
                    System.out.println("接收市场报表："+msg);
                }
            }