网站cms模板,手机微信网页版,WordPress最好的免费主题,诊断网站seo现状前边我们有具体的学习过RabbitMQ的安装和基本使用的情况。 但是呢#xff0c;没有演示具体应用到项目中的实例。 这里使用RabbitMQ来实现流量的削峰添谷。 一#xff1a;添加pom依赖
!--rabbitmq-需要的 AMQP 依赖--
dependencygroupIdorg.springfr…前边我们有具体的学习过RabbitMQ的安装和基本使用的情况。 但是呢没有演示具体应用到项目中的实例。 这里使用RabbitMQ来实现流量的削峰添谷。 一添加pom依赖
!--rabbitmq-需要的 AMQP 依赖--
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency 二yml配置
spring:
#配置rabbitmq 服务器
rabbitmq:virtual-host: /host: 1.15.157.156port: 5672username: xxxxxpassword: xxxxx# 开启发布确认机制#SIMPLE, // 使用 RabbitTemplate#waitForConfirms() 或 waitForConfirmsOrDie()#CORRELATED, // 使用 CorrelationData 关联确认与发送的消息#NONE // 不启用发布确认publisher-confirm-type: correlated# publisher-confirms 消息的可靠投递, confirm 确认模式 默认为false#publisher-confirms: true# 添加发布确认返回, return 回退模式 默认为falsepublisher-returns: true### listenerlistener:# 每次从队列中预取5条消息prefetch: 20# 最小消费者数量concurrency: 1# 最大的消费者数量max-concurrency: 10simple:# 设置预取数量为1 每次取一个prefetch: 1# manual手动 ack需要在业务代码结束后调用 api 发送 ack虽灵活但会提高编码复杂度。# auto自动 ack没有异常则返回 ack抛出异常则返回 nack消息重新入队一直到没有异常为止也可以设置最大重试次数超过次数后发送到专门收集错误消息的队列进一步处理# none关闭ackMQ 假定消费者获取消息后会成功处理因此消息投递后立即被删除消息投递是不可靠的可能丢失acknowledge-mode: manual# 失败重试retry:# 开启消费者失败重试enabled: true# 初始的失败等待时长为1秒initial-interval: 1000# 失败的等待时长倍数下次等待时长 multiplier * last-intervalmultiplier: 3# 最大重试次数max-attempts: 4# true无状态false有状态。如果业务中包含事务这里改为falsestateless: true
具体的配置都有对应的注释参照即可。 三编写config配置类
package com.modules.config;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;Configuration
public class RabbitMQConfig
{Value(${spring.rabbitmq.host})private String host;Value(${spring.rabbitmq.port})private int port;Value(${spring.rabbitmq.username})private String userName;Value(${spring.rabbitmq.password})private String password;Value(${spring.rabbitmq.listener.prefetch})private int prefetch;Value(${spring.rabbitmq.listener.concurrency})private int concurrentConsumers;Value(${spring.rabbitmq.listener.max-concurrency})private int maxConcurrentConsumers;/*** 链接RabbitMQ* return*/Beanpublic ConnectionFactory connectionDirectFactory(){CachingConnectionFactory connectionFactory new CachingConnectionFactory();connectionFactory.setHost(host);connectionFactory.setPort(port);connectionFactory.setUsername(userName);connectionFactory.setPassword(password);connectionFactory.setPublisherConfirms(true); //必须要设置return connectionFactory;}/*** 配置RabbitMQ参数* return*/Beanpublic SimpleRabbitListenerContainerFactory rabbitDirectListenerContainerFactory(){SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionDirectFactory());//设置最小并发的消费者数量factory.setConcurrentConsumers(concurrentConsumers);//设置最大并发的消费者数量factory.setMaxConcurrentConsumers(maxConcurrentConsumers);//限流单位时间内消费多少条记录factory.setPrefetchCount(prefetch);// json转消息//factory.setMessageConverter(new Jackson2JsonMessageConverter());//设置rabbit 确认消息的模式默认是自动确认//factory.setAcknowledgeMode(AcknowledgeMode.AUTO);//设置rabbit 确认消息的模式默认是自动确认factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);return factory;}/*** 回调函数* param connectionFactory* return*/Beanpublic RabbitTemplate createDirectRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//设置开启Manatory才能触发回调函数无论消息推送结果怎么样都会强制调用回调函数rabbitTemplate.setMandatory(true);// 设置确认发送到交换机的回调函数 》 消息推送到server但是在server里找不到交换机 / 消息推送到sever交换机和队列啥都没找到 / 消息推送到server找到交换机了但是没找到队列 / 消息推送成功rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - {if(ack){System.out.println(发送者消息确认成功);}else{System.out.println(发送者消息确认是呗考虑重发cause);}//System.out.println(相关数据correlationData);//System.out.println(确认情况ack);//System.out.println(原因cause);//System.out.println();});//设置确认消息已发送到队列的回调 》 消息推送到server找到交换机了但是没找到队列 触发这个回调函数rabbitTemplate.setReturnsCallback(returnedMessage - {System.out.println(交换机为returnedMessage.getExchange());System.out.println(返回消息为returnedMessage.getMessage());System.out.println(路由键为returnedMessage.getRoutingKey());System.out.println(回应消息为returnedMessage.getReplyText());System.out.println(回应代码为returnedMessage.getReplyCode());System.out.println();});return rabbitTemplate;}BeanQueue trafficSpikedQueue(){return new Queue(trafficSpikedQueue, true);}BeanDirectExchange trafficSpikedExchange(){return new DirectExchange(trafficSpikedExchange);}BeanBinding binding(Queue trafficSpikedQueue, DirectExchange trafficSpikedExchange){return BindingBuilder.bind(trafficSpikedQueue).to(trafficSpikedExchange).with(trafficSpikedKey);}//*/
} 四创建生产者
package com.modules.controller.rabbitmq;import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;RestController
public class TrafficController {Autowiredprivate RabbitTemplate rabbitTemplate;GetMapping(/java/traffic)public String sendTrafficMessage(RequestParam String message){for (int i 1; i 100; i){// 使用java多线程来模拟多用户并发请求final int temp i;new Thread(()-{// 给RabbitMQ发送消息rabbitTemplate.convertAndSend(trafficExchange,trafficKey,hello world:temp,new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException{// System.out.println(发送回调:temp);System.out.println(message);return message;}});}).start();}// rabbitTemplate.convertAndSend(trafficSpikedExchange, trafficSpikedKey, message);return Message sent;}
} 五创建消费者
package com.modules.controller.rabbitmq;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.amqp.core.Message;
import com.rabbitmq.client.*;
import java.io.IOException;Component
public class TrafficSpikedConsumer {RabbitListener(queues trafficQueue)public void receiveMessage(Message message, Channel channel) throws InterruptedException, IOException{// 为了演示一个一个消费的情况这里使用线程暂停来延迟控制台输出Thread.sleep(100);// // 处理消息例如写入数据库或进行计算System.out.println(Received message: new String(message.getBody()));//System.out.println(channel: channel);// // 成功处理后手动确认消息long deliveryTag message.getMessageProperties().getDeliveryTag();//System.out.println(deliveryTag:deliveryTag);channel.basicAck(deliveryTag, false);}
}
控制台输出的数据比较多。我这里就不做展示了。 PS我这里测试的时候遇到一个小问题发现消费者最后消费的数量跟生产者生产的数量对不上。我百思不得其解。这问题出在哪里呢
后来我才发现我测试是在本地做的测试对应的代码我服务器端打包的jar里边也有一份也就是说我一个生产者对应两个消费者本地服务器这也是我本地消费者消费的数量跟生产数量不一致的原因。 以上大概就是Springboot集成RabbitMQ实现流量削峰添谷的一个小例子。 通过RabbitMQ的队列机制可以有效地缓解高峰期的流量压力。 有好的建议请在下方输入你的评论。
