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

网站开发语言比例,南通公司注册,北京建站模板厂家,app制作过程/*** 使用rabbitMQ* 1.引用amqp场景 RabbitAutoConfiguration就会自动生效* 2.给容器中自动配置了各种api RabbitTemplate AmqpAdmin CachingConnectionFactory RabbitMessagingTemplate* 所有属性都是 spring.rabbitmq开头* 3.通过注解EnableRabbit使用* 4.监听消息 使用Rabbi…/*** 使用rabbitMQ* 1.引用amqp场景 RabbitAutoConfiguration就会自动生效* 2.给容器中自动配置了各种api RabbitTemplate AmqpAdmin CachingConnectionFactory RabbitMessagingTemplate* 所有属性都是 spring.rabbitmq开头* 3.通过注解EnableRabbit使用* 4.监听消息 使用RabbitListener 注解 必须有EnableRabbit才能生效 如果是创建交换机创建队列 不需要有EnableRabbit注解* RabbitListener 可以标在类和方法上* RabbitHandler 可以标在方法上 场景 一个队列返回的类型不同 使用这个注解来重载*/ 1.引入依赖 !– mq依赖–dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency 2.配置文件 spring.rabbitmq.host127.0.0.1 spring.rabbitmq.port5672 #虚拟主机 spring.rabbitmq.virtual-host/#开启消息生产者发送消息确认

NONE:禁用发布确认模式是默认值

#CORRELATED:发布消息成功到交换器后会触发回调方法 #SIMPLEspring.rabbitmq.publisher-confirm-typecorrelated

开启发送端消息抵达队列的确认

spring.rabbitmq.publisher-returnstrue

只要抵达队列以异步发送优先回调这个

spring.rabbitmq.template.mandatorytrue#手动ack消息 spring.rabbitmq.listener.simple.acknowledge-modemanual 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 MapString, 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 MapString, 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 i0;i10;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 ServiceImplOrderItemDao, 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(correlationDatacorrelationDatabbss);}});//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机制进行手动确认 常见的模式 1.点对点模式如果路由key和绑定关系完全匹配交换机才能收到 2.主题订阅模式路由key和绑定关系以单词维度匹配路由key中可以用  “#”代表匹配多个或0个 “” 代表匹配一个 3.广播模式不受路由key局限只要交换机和队列有绑定关系就可以收到消息 4.helder性能比较低一般不适用私信队列不太熟悉 rabbitmq的基本原理