RabbitMQ的安装、配置和实战

RabbitMQ安装

1. docker run -d  --name xd_rabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=password -p 15672:15672 -p 5672:5672 rabbitmq:3.8.15-management
4. #网络安全组记得开放端口
5. 4369 erlang 发现口
6. 5672 client 端通信口
7. 15672 管理界面 ui 端口
8. 25672 server 间内部通信口
10. 访问管理界面
11. ip:15672

复制代码

依赖引入

1. <dependency>
2.     <groupId>org.springframework.boot</groupId>
3.     <artifactId>spring-boot-starter-amqp</artifactId>
4. </dependency>

复制代码

application.xml

1. spring:
2.   ##----------rabbit配置--------------
3.   rabbitmq:
4.     host: 192.168.75.146
5.     port: 5672
6.     virtual-host: dev
7.     username: admin
8.     password: password
9.     listener:
10.       simple:
11.         #消息确认方式，manual(手动ack) 和auto(自动ack) 。消息队列重试到达次数进入异常交换机--为实现，该策略要为auto
12.         acknowledge-mode: auto
13.         retry:
14.           #开启重试，消费者代码不能添加try catch捕获不往外抛异常
15.           enabled: true
16.           #最大重试次数
17.           max-attempts: 4
18.           # 重试消息的时间间隔，5秒
19.           max-interval: 5000

复制代码

RabbitMQ配置文件 (一个交换机，两个队列，routingKey匹配规则适用于两队列)

1. @Configuration
2. @Data
3. public class RabbitMQConfig {
6.     /**
7.      * 交换机
8.      */
9.     private String shortLinkEventExchange="short_link.event.exchange";
11.     /**
12.      * 创建交换机 Topic类型
13.      * 一般一个微服务一个交换机
14.      * @return
15.      */
16.     @Bean
17.     public Exchange shortLinkEventExchange(){
19.         return new TopicExchange(shortLinkEventExchange,true,false);
20.         //return new FanoutExchange(shortLinkEventExchange,true,false);
21.     }
23.     //新增短链相关配置====================================
25.     /**
26.      * 新增短链 队列
27.      */
28.     private String shortLinkAddLinkQueue="short_link.add.link.queue";
30.     /**
31.      * 新增短链映射 队列
32.      */
33.     private String shortLinkAddMappingQueue="short_link.add.mapping.queue";
35.     /**
36.      * 新增短链具体的routingKey,【发送消息使用】
37.      */
38.     private String shortLinkAddRoutingKey="short_link.add.link.mapping.routing.key";
40.     /**
41.      * topic类型的binding key，用于绑定队列和交换机，是用于 link 消费者
42.      */
43.     private String shortLinkAddLinkBindingKey="short_link.add.link.*.routing.key";
45.     /**
46.      * topic类型的binding key，用于绑定队列和交换机，是用于 mapping 消费者
47.      */
48.     private String shortLinkAddMappingBindingKey="short_link.add.*.mapping.routing.key";
51.     /**
52.      * 新增短链api队列和交换机的绑定关系建立
53.      */
54.     @Bean
55.     public Binding shortLinkAddApiBinding(){
56.         return new Binding(shortLinkAddLinkQueue,Binding.DestinationType.QUEUE, shortLinkEventExchange,shortLinkAddLinkBindingKey,null);
57.     }
60.     /**
61.      * 新增短链mapping队列和交换机的绑定关系建立
62.      */
63.     @Bean
64.     public Binding shortLinkAddMappingBinding(){
65.         return new Binding(shortLinkAddMappingQueue,Binding.DestinationType.QUEUE, shortLinkEventExchange,shortLinkAddMappingBindingKey,null);
66.     }
69.     /**
70.      * 新增短链Link普通队列，用于被监听
71.      */
72.     @Bean
73.     public Queue shortLinkAddLinkQueue(){
75.         return new Queue(shortLinkAddLinkQueue,true,false,false);
77.     }
79.     /**
80.      * 新增短链mapping 普通队列，用于被监听
81.      */
82.     @Bean
83.     public Queue shortLinkAddMappingQueue(){
85.         return new Queue(shortLinkAddMappingQueue,true,false,false);
87.     }

复制代码

对应的两个消费者

1. @Component
2. @Slf4j
3. //@RabbitListener(queues = "short_link.add.link.queue")
4. @RabbitListener(queuesToDeclare = { @Queue("short_link.add.link.queue") })
5. public class ShortLinkAddLinkMQListener {
7.     @Autowired
8.     private ShortLinkService shortLinkService;
10.     @RabbitHandler
11.     public void shortLinkHandler(EventMessage eventMessage, Message message, Channel channel) throws IOException {
12.         log.info("ShortLinkAddLinkMQListener message:{}",message);
13.         try{
14.             eventMessage.setEventMessageType(EventMessageType.SHORT_LINK_ADD_LINK.name());
15.             shortLinkService.handlerAddShortLink(eventMessage);
17.         }catch (Exception e){
18.             // 处理业务异常，等其他操作
19.             log.error("消费失败:{}",eventMessage);
20.             throw new BizException(BizCodeEnum.MQ_CONSUME_EXCEPTION);
21.         }
22.         log.info("消费成功:{}",eventMessage);
23.         //确认消息消费成功
24. //        channel.basicAck(tag,false);
26.     }
27. }

复制代码

1. @Component
2. @Slf4j
3. //@RabbitListener(queues = "short_link.add.mapping.queue")
4. @RabbitListener(queuesToDeclare = { @Queue("short_link.add.mapping.queue") })
5. public class ShortLinkAddMappingMQListener {
6.     @Autowired
7.     private ShortLinkService shortLinkService;
9.     @RabbitHandler
10.     public void shortLinkHandler(EventMessage eventMessage, Message message, Channel channel) throws IOException {
11.         log.info("ShortLinkAddMappingMQListener message:{}",message);
12.         try{
13.             eventMessage.setEventMessageType(EventMessageType.SHORT_LINK_ADD_MAPPING.name());
14.             shortLinkService.handlerAddShortLink(eventMessage);
16.         }catch (Exception e){
17.             // 处理业务异常，等其他操作
18.             log.error("消费失败:{}",eventMessage);
19.             throw new BizException(BizCodeEnum.MQ_CONSUME_EXCEPTION);
20.         }
21.         log.info("消费成功:{}",eventMessage);
22.         //确认消息消费成功
23. //        channel.basicAck(tag,false);
25.     }
26. }

复制代码

实战，这里直接在controller层中编码。（应该在Service实现）

1. @RestController
2. @RequestMapping("/api/link/v1")
3. public class ShortLinkController {
5.     @Autowired
6.     private ShortLinkService shortLinkService;
7.    
8.     @Autowired
9.     private RabbitTemplate rabbitTemplate;
10.    
11.     /**
12.      * 新增短链
13.      * @param shortLinkAddRequest
14.      * @return
15.      */
16.     @PostMapping("add")
17.     public JsonData createShortLink(@RequestBody ShortLinkAddRequest shortLinkAddRequest){
18.         //参数：交换机、匹配规则Key、信息对象
19.         EventMessage eventMessage = EventMessage.builder().[设置对象各字段信息]
20.             .build();
22.         //生成MQ。
23.         rabbitTemplate.convertAndSend(rabbitMQConfig.getShortLinkEventExchange(), rabbitMQConfig.getShortLinkAddRoutingKey(), eventMessage);
24.         
25.         return jsonData;
26.     }
27.    
28. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！