当前位置：首页 > news >正文

RabbitMQ的基础使用

news 2025/9/17 5:24:29

/*** 使用rabbitMQ* 1.引用amqp场景 RabbitAutoConfiguration就会自动生效* 2.给容器中自动配置了各种api RabbitTemplate AmqpAdmin CachingConnectionFactory RabbitMessagingTemplate* 所有属性都是 spring.rabbitmq开头* 3.通过注解@EnableRabbit使用* 4.监听消息 使用@RabbitListener 注解 必须有@EnableRabbit才能生效 如果是创建交换机，创建队列 不需要有@EnableRabbit注解* @RabbitListener 可以标在类和方法上* @RabbitHandler 可以标在方法上 场景 一个队列返回的类型不同 使用这个注解来重载*/

1.引入依赖

<!--        mq依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

2.配置文件

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
#虚拟主机
spring.rabbitmq.virtual-host=/#开启消息生产者发送消息确认
# NONE:禁用发布确认模式，是默认值
#CORRELATED:发布消息成功到交换器后会触发回调方法
#SIMPLEspring.rabbitmq.publisher-confirm-type=correlated
# 开启发送端消息抵达队列的确认
spring.rabbitmq.publisher-returns=true
# 只要抵达队列，以异步发送优先回调这个
spring.rabbitmq.template.mandatory=true#手动ack消息
spring.rabbitmq.listener.simple.acknowledge-mode=manual

3.注入

 @AutowiredAmqpAdmin amqpAdmin;

操作创建交换机，创建队列，创建绑定关系

 /*** 1.如何创建Exchange    Queue Binding?  使用AmqpAdmin进行创建* 2.如何接收消息*/@Testpublic void createExchange(){//创建一个交换机//String name  交换机名称, boolean durable 是否持久化, boolean autoDelete  是否自动删除DirectExchange directExchange = new DirectExchange("hallo-java-exchange", true, false);amqpAdmin.declareExchange(directExchange);System.out.println("单点交换机创建成功"+directExchange);}@Testpublic void createQueue(){//String name 队列名称, boolean durable 是否持久化, boolean exclusive 是否排他,// boolean autoDelete 是否自动删除, @Nullable Map<String, Object> argumentsQueue queue = new Queue("hello-java-queue",true,false,false);amqpAdmin.declareQueue(queue);}/*** 测试创建绑定关系*/@Testpublic void createBing(){//String destination, 目的地// DestinationType destinationType,目的地类型//String exchange 交换机// String exchange, String routingKey, 路由key// @Nullable Map<String, Object> arguments 自定义参数Binding binding = new Binding("hello-java-queue", Binding.DestinationType.QUEUE,"hallo-java-exchange","hello.java",null);amqpAdmin.declareBinding(binding);}

/ 用来发送消息的@AutowiredRabbitTemplate rabbitTemplate;

/*** 测试发送消息功能*/@Testpublic void sendMessageTest(){for(int i=0;i<10;i++){if(i%2 == 0){OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();reasonEntity.setId(1L);reasonEntity.setCreateTime(new Date());reasonEntity.setName("哈哈");//发送消息 如果发送的消息是个对象，我们会使用序列化机制，将对象写出去 对象必须实现 Serializable//也可以用config的方法 将对象类型的消息转为json//String exchange 发送的交换机, String routingKey 路由key, Object object 发送的内容,// correlationData 指定uuid,生产者发送消息给服务时候，获得这个参数，从而确定是哪条消息rabbitTemplate.convertAndSend("hallo-java-exchange","hello.java",reasonEntity,new CorrelationData(UUID.randomUUID().toString()));}else{OrderEntity orderEntity = new OrderEntity();orderEntity.setOrderSn(UUID.randomUUID().toString());rabbitTemplate.convertAndSend("hallo-java-exchange","hello.java",orderEntity,new CorrelationData(UUID.randomUUID().toString()));}}}

接收消息在业务层操作相关的注解有

@RabbitListener 可以标在类和方法上

@RabbitHandler 可以标在方法上场景一个队列返回的类型不同使用这个注解来重载

@Service("orderItemService")
@RabbitListener(queues = {"hello-java-queue"})
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {/*** 测试接收消息* 注解中的queues 声明需要监听的队列，可以是多个* org.springframework.amqp.core.Message 第一个参数是原生的详细信息* OrderReturnApplyEntity 第二个参数 消息的返回类型* Channel 第三个参数 服务获取消息的通道** 场景* 多个客户端启动时，一个消息只能被一个客户端接收 轮询制度* 多个消息等待服务处理时，处理完一个后，才会接收下一个** 设置ack配置 确认前是unack状态 若宕机或其他原因失败，系统再次重启状态为ready 手动签收后消息才被处理掉**/@RabbitHandlerpublic void recieveMessage(Message message, OrderReturnApplyEntity returnApplyEntity,Channel channel) throws IOException {//获取消息体byte[] body = message.getBody();//获取消息头MessageProperties messageProperties = message.getMessageProperties();System.out.println(returnApplyEntity);long deliveryTag = message.getMessageProperties().getDeliveryTag();//channel内按照顺自增的if(deliveryTag%2 == 0){//当业务流程执行完后 手动签收消息 deliveryTag 签收的消息序号, b 是否批量签收channel.basicAck(deliveryTag,false);}else{//拒签 拒签的序号，是否批量操作，是否归队 若false则丢弃 若ture则归队channel.basicNack(deliveryTag,false,false);// 和上面作用一样 只是不能设置批量操作这个参数// channel.basicReject(deliveryTag,false);}}@RabbitHandlerpublic void recieveMessage2(Message message, OrderEntity order,Channel channel){//获取消息体byte[] body = message.getBody();//获取消息头MessageProperties messageProperties = message.getMessageProperties();System.out.println(order);}

4.config文件

@Configuration
public class MyRabbitConfig {@AutowiredRabbitTemplate rabbitTemplate;@Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}/*** 定制RabbitTemplate*/@PostConstruct //MyRabbitConfig对象创建完成后，执行这个方法public void initRabbitTemplate(){//设置确认回调//1.生产者发送消息到服务这一步的确认rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {//CorrelationData correlationData, 当前消息的唯一关联数据（可以理解为id）// boolean b, 判断消息是否成功收到// String s 如果失败，显示失败原因@Overridepublic void confirm(CorrelationData correlationData, boolean b, String s) {System.out.println("correlationData==="+correlationData+"b==="+b+"s==="+s);}});//2.设置消息没有投递给指定的队列，就触发这个失败的回调rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {//获得投递失败的消息的详细信息Message message = returnedMessage.getMessage();//获得投递消息失败的交换机String exchange = returnedMessage.getExchange();//获得投递消息失败的状态码int replyCode = returnedMessage.getReplyCode();//获得投递消息失败的文本String replyText = returnedMessage.getReplyText();//获得投递消息失败的路由keyString routingKey = returnedMessage.getRoutingKey();}});//3.消费端确认（保证每个消息 被正确消费，此时才可以删除这个消息）}

保证消息可靠投递

生产者到交换机使用confirmCallback; 得到消息投递的结果

交换机到队列使用 returnCallback; 如果消息接收失败，将会触发，得到失败消息的信息

队列到消费者使用ack机制；进行手动确认