RbbitMQ与springboot的整合

傷城~ 2022-11-27 12:16 174阅读 0赞

1 引入依赖

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

2 YML中的配置

spring:
      rabbitmq:
        host: 192.168.1.245
        port: 5672
        username: admin
        password: admin
        virtual-host: my_vhost
        listener:
          simple:
            concurrency: 1 #最小消费者数量
            max-concurrency: 1 #最大消费者数量
            prefetch: 1 #在单个请求中处理的消息个数
            acknowledge-mode: auto  #消费端手动应答或者自动应答，这里设置自动应答

3 发布者发布消息

@Slf4j
    @RestController
    @RequestMapping("/rabbit")
    public class RabbitMqController { 
    
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        /** * 简单模式Hello World */
        @GetMapping("/hello")
        public void testHello(){ 
            rabbitTemplate.convertAndSend("hello","hello world");
            System.out.println("消息发送完毕");
        }
    
    
        @GetMapping("/work")
        public void testWork(){ 
            rabbitTemplate.convertAndSend("work","多个消费者消费");
            System.out.println("消息发送完毕");
        }
    
        /** * 广播模型 */
        @GetMapping("/fanout")
        public void testFanout(){ 
    
            rabbitTemplate.convertAndSend("logs","","广播消息");
            System.out.println("消息发送完毕");
        }
    
        /** * 路由模型 */
        @GetMapping("/route")
        public void testRoute(){ 
    
            rabbitTemplate.convertAndSend("logs","info","广播info日志");
            rabbitTemplate.convertAndSend("logs","warn","广播warn日志");
            rabbitTemplate.convertAndSend("logs","error","广播error日志");
            System.out.println("消息发送完毕");
        }
    
        /** * 订阅模型 */
        @GetMapping("/topic")
        public void testTopic(){ 
    
            rabbitTemplate.convertAndSend("logs_Topics","china.weather","今天中国天气真好");
            rabbitTemplate.convertAndSend("logs_Topics","china.news","今天中国中芯国际上市了");
            rabbitTemplate.convertAndSend("logs_Topics","usa.weather","今天美国天气非常差");
            rabbitTemplate.convertAndSend("logs_Topics","usa.news","今天美国疫情突破两百万");
            System.out.println("消息发送完毕");
        }

4 消费者  
4.1 简单模式Hello World

@Component
    @RabbitListener(queuesToDeclare = @Queue("hello"))
    public class HelloCustomer { 
    
        @RabbitHandler
        public void receive(String msg){ 
            System.out.println("消息已经收到 ="+msg);
        }
    }

4.2 work模型

@Component
    public class WorkCustomer { 
    
        @RabbitListener(queuesToDeclare = @Queue("work"))
        public void receive1(String msg){ 
            System.out.println("消息已经收到1 ="+msg);
        }
    
        @RabbitListener(queuesToDeclare = @Queue("work"))
        public void receive2(String msg){ 
            System.out.println("消息已经收到2 ="+msg);
        }
    }

4.3 广播模型

@Component
    public class FanoutCustomer { 
    
        @RabbitListener(bindings = { 
                @QueueBinding(value = @Queue,//临时队列
                        exchange = @Exchange(value = "logs", type = ExchangeTypes.FANOUT)
                )}
        )
        public void receive1(String msg) { 
            System.out.println("消息已经收到1 =" + msg);
        }
    
        @RabbitListener(bindings = { 
                @QueueBinding(value = @Queue,//临时队列
                        exchange = @Exchange(value = "logs", type = ExchangeTypes.FANOUT)
                )}
        )
        public void receive2(String msg) { 
            System.out.println("消息已经收到2 =" + msg);
        }
    }

4.4 路由模型

@Component
    public class RouteCustomer { 
    
        @RabbitListener(bindings = { 
                @QueueBinding(value = @Queue,//临时队列
                        exchange = @Exchange(value = "logs", type = ExchangeTypes.DIRECT),
                        key = { "info","warn"}
                )}
        )
        public void receive1(String msg) { 
            System.out.println("消息已经收到1 =" + msg);
        }
    
        @RabbitListener(bindings = { 
                @QueueBinding(value = @Queue,//临时队列
                        exchange = @Exchange(value = "logs", type = ExchangeTypes.DIRECT),
                        key = { "info","error"}
                )}
        )
        public void receive2(String msg) { 
            System.out.println("消息已经收到2 =" + msg);
        }
    }

4.5 订阅模型

@Component
    public class TopicCustomer { 
    
        @RabbitListener(bindings = { 
                @QueueBinding(value = @Queue,//临时队列
                        exchange = @Exchange(value = "logs", type = ExchangeTypes.TOPIC),
                        key = { "china.*"}
                )}
        )
        public void receive1(String msg) { 
            System.out.println("消息已经收到1 =" + msg);
        }
    
        @RabbitListener(bindings = { 
                @QueueBinding(value = @Queue,//临时队列
                        exchange = @Exchange(value = "logs", type = ExchangeTypes.TOPIC),
                        key = { "usa.*"}
                )}
        )
        public void receive2(String msg) { 
            System.out.println("消息已经收到2 =" + msg);
        }
    }