RabbitMq学习笔记(3)_SpringBoot整合RabbitMQ

时间:2020-6-28 作者:admin


七.Boot整合RabbitMQ

整合思路:
  1. 引入 spring-boot-starter-amqp
  2. 2.application.yml配置
  3. 测试RabbitMQ

​ AmqpAdmin:管理组件

​ RabbitTemplate:消息发送处理组件

1. 创建boot工程,加入rabbitmq启动器
2. 在配置文件中加入rabbintmq 的配置信息
spring.rabbitmq.addresses=192.168.50.128
spring.rabbitmq.username=guest       
spring.rabbitmq.password=guest       
#spring.rabbitmq.virtual-host=/      
3. 使用RabbitTemplate操作rabbitMq
RabbitAutoConfiguration,ConnectionFactory,RabbitProperties自动配置;
注入RabbitTemplate:给rabbitmq发送和接收消息
AmqpAdmin: RabbitMq系统管理功能(声明队列,创建交换器)
4. 测试RabbitMQ的发送和接收消息
@RunWith(SpringRunner.class)
	@SpringBootTest
	public class AmqpApplicationTests {
   	//注入template模版
  	 @Autowired
  	 RabbitTemplate rt;
  	 @Test
   	public void contextLoads() {
      	//点对点发送
      	rt.convertAndSend("exchanges.direct","oracle","我是一只小小的石头,i am a small small stone");
      	//向fanout发送消息
      	rt.convertAndSend("exchanges.fanout","oracle","我是一只小小的fanout,i am a small small fanout");
      	//给topic发送消息
     	 rt.convertAndSend("exchanges.topic","oracle.abc","我是一只小小的topic,i am a small small topic");
   	}
	
  	 @Test
   	public void receive(){
      	//接收队列的消息
     	 Object obj=rt.receiveAndConvert("oracle");
      	System.out.println(obj);
   	}
    }
/*定义一个发消息的Service*/
package com.oracle.messageService;

import com.oracle.vo.Book;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderMessage {
    @Autowired
    RabbitTemplate rabbitTemplate;

    public void send(String message){
       Object obj= rabbitTemplate.convertSendAndReceive("amq.direct","oracle",message);
        System.out.println("结果是:"+obj);
    }


    public void send(Book book){
        Object obj= rabbitTemplate.convertSendAndReceive("order.topic","oracle.news",book);
        System.out.println("结果是:"+obj);
    }
    public void receive(){
        Object obj=rabbitTemplate.receiveAndConvert("oracle.over");
        System.out.println("obj:"+obj);
    }
}

5.配置消息转换器

打开RabbitTemplate源码后,找到MessageConverter类,然后在IDEA中使用ctrl+H查看到继承关系后,找到

Jackson2JsonMessageConverter;并在配置文件中进行配置;


@Configuration
public class MessageConvertor {

    @Bean
    public MessageConverter jacksonMessageConvertor(){
        Jackson2JsonMessageConverter jackson2JsonMessageConverter=new Jackson2JsonMessageConverter();
        return  jackson2JsonMessageConverter;
    }
}
注:如果配置了消息转换器,那么使用@RabbitListener也需要配置相同的转换器,否则会转换失败;
6.** 编写service监听队列

​ @RabbitListener(queues = “oracle”)
​ 此注解放在方法上,监听队列

  1. 启动类
	@EnableRabbit //启动rabbit
	@SpringBootApplication
	public class AmqpApplication {
 		public static void main(String[] args) {
   			SpringApplication.run(AmqpApplication.class, args);
  		}
	}
  1. service类
	@Service
	public class RabbitService {
  

	@RabbitListener(queues = "oracle")
  	public void oracle(String str) {
    		System.out.println("收到了oracle队列中的 消息:" + str);
  	}

  	@RabbitListener(queues = "book")
  	public void book(Book book){
    		System.out.println("收到一本书: "+book);
   }
}

  1. 配置类

    spring.rabbitmq.addresses=127.0.0.1
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=tiger
    spring.rabbitmq.virtual-host=order
    
7.特殊说明

消息接收者和消息发送查应该使用相同的消息转换器;默认是字节序列化。如果要改成json方式的,发送与接收到应该配置json消息转换器;

8.管理类;
@Autowired
AmqpAdmin amqpAdmin;

@Test
public void create(){
  //创建一个exchange
  amqpAdmin.declareExchange(new DirectExchange("amqp.direct"));
  //创建一个队列
  amqpAdmin.declareQueue(new Queue("amqp.test2"));
  //绑定交换器
  amqpAdmin.declareBinding(new Binding("amqp.test", Binding.DestinationType.QUEUE,"amqp.direct","qmqp.test",null));
}

Bingding构造方法的参数:

Amqp.test:队列名

Binding.DestinationType.QUEUE:amqp.test的类型是队列

Amqp.direct:exchange名称

Qmqp.test:路由键

Null:叁数

操作步骤:

  1. 每执行完一个操作后,都去后台管理去查看一下操作后的结果;
  2. 创建对象使用declaredXXX,删除对象使用deleteXXX
  3. 通过查看源码的方式来操作,使用ctrl+n来查找类或接口,再使用ctrl+h查看类的层次结构;

八.RabbitMQ中的消息确认ACK机制

什么是消息确认ACK?

​ 如果在处理消息的过程中,消费者的服务器在处理消息时出现异常,那可能这条正在处理的消息就没有完成消息消费,数据就会丢失。为了确保数据不丢失,RabbitMQ支持消息确认,即ACK

ACK的消息确认机制

​ ACK机制是消费者从RabbitMQ收到消息并处理完成后,反馈给RabbitMQ,RabbitMQ收到反馈后才将次消息从队列中删除

  1. 如果一个消费者在处理消息出现了网络不稳定、服务器异常等现象,那么就不会有ACK反馈,RabbitMQ会认为这个消息没有正常消费,会将消息重新放入队列中
  2. 如果在集群的情况下:RabbitMQ会立刻将这个消息推送给这个在线的其他消费者。这种机制保证了在消费者服务端故障的时候,不丢失任何消息和任务。
  3. **消息永远不会从RabbitMQ中删除:**只有当消费者正确返送ACK反馈,RabbitMQ确认收到后,消息才会从RabbitMQ服务的数据中删除。
  4. 消息的ACK确认机制默认是打开的。

ACK机制的开发注意实现

​ 如果忘记了ACK,那么后果很严重。当Consumer退出时,Message会一直重新分发。然后RabbitMQ会占用越来越多的内存,由于RabbitMQ会长时间运行,因此这个“内存泄漏”是致命的。

练习:

​ 创建两个模块,消费者和生产者使用direct交换器,在消费者消费消息时抛出一个RuntimeException,观察控制台和RabbitMQ的Web控制台,消息会再次重新放到队列中,执行死循环。

receiver:

@Component
@RabbitListener(bindings=@QueueBinding(
        value=@Queue(value="${mq.config.queue.info}",autoDelete = "true"),
        exchange = @Exchange(value="${mq.config.exchange}",type= ExchangeTypes.DIRECT),
        key = "${mq.config.queue.info.routing.key}"
))
public class InfoReceiver {
    @RabbitHandler
    public void process(String msg){
        System.out.println("---------consumer:receive:"+msg);
        throw new RuntimeException();
    }
}

RabbitMQ web控制台

RabbitMq学习笔记(3)_SpringBoot整合RabbitMQ

解决方式:

方式一:可以在消费消息中try-catch

​ 由于在消费消息时可能抛出运行时异常,这种解决方式难于处理和控制。

方式二:在Consumer的配置文件中配置重试机制

​ 配置开启重试及重试次数,在RabbitMQ没有收到ACK确认时,重新放入队列,重试指定次数后不再放入队列。

spring.application.name=mq-direct-consumer
spring.rabbitmq.host=192.168.37.151
spring.rabbitmq.port=5672
spring.rabbitmq.username=testww
spring.rabbitmq.password=000000

#开启重试机制
spring.rabbitmq.listener.simple.retry.enabled=true
#重试次数 默认为3次 现修改为5次
spring.rabbitmq.listener.simple.retry.max-attempts=5
声明:本文内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:87074139@qq.com 进行举报,并提供相关证据,工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。