淄博市建设档案馆网站,wordpress uc点赞,友情链接有哪些作用,成都网站设公司什么是消息中间件 简单的来说就是消息队列中间件#xff0c;生产者发送消息到中间件#xff0c;消息中间件用于 保存消息并发送消息到消费者。 消息中间件RabbitMQ的基本组件 1#xff09;producer -生产者 2#xff09;customer -消费者 3#xff09;broker (经纪人)- M… 什么是消息中间件 简单的来说就是消息队列中间件生产者发送消息到中间件消息中间件用于 保存消息并发送消息到消费者。 消息中间件RabbitMQ的基本组件 1producer -生产者 2customer -消费者 3broker (经纪人)- MQ服务器管理消息对列、消息及相关消息。接收并存储生产者发送的消息发送消息到消费者 4exchange-交换机将生产者的消息按照一定规则发送给对应的消息对列queue 5queue-消息对列队列消息存放的容器消息先进先出 6Message-消息程序间的通信的数据 什么是消息队列queue(生产者生产msg-queue消费者监听queue-消费) 消息对列是一种分布式中的通信方式它通过异步传输消息的方式来解耦消 息的 生产者和消费者。在消息中间件中生产者将消息发送到消息对列中以为先进先出的方式消费者从对列中取出消息可以监听对列是否有消息-RabblitListener和RabbitHandler 消息中间件的作用 主要有三个作用分别是服务解耦、实现异步通信、流量削峰 1. 服务解耦场景-用户下订单、库存服务工作 例如订单服务-用户下订单库存服务处理对应减库存才返回给用户下单成功的消息。如果说库存服务出现了问题就会造成订单丢失等问题。如果使用消息中间件消息对列可以把下的订单信息— mq就返回用户下单这个mq再发送给库存服务这样生产者发送消息和消费者接收处理消息相互不影响即使宕机了消息还在中间件中。
2). 异步通信/异步调用用户注册新用户服务发送短信和邮件 传统的模式用户注册系统新用户服务给用户发送短信和邮件三个操作都完成之后才返回用户下单注册的消息。因为短信和邮箱和注册信息是没有关系的服务用户注册后消息发送给mq用户不需要等邮件和短信发送成功mq直接返回用户注册成功至此用户注册业务完成。至于短信和邮件交给mq发送给短信业务-去发送。
注意 异步就是某线程发出请求不需要等其他线程完成就接着完成操作。用户注册消息发送给mq不需要等短信服务完成短信发布发送都与注册无关两者是异步关系。异步不是并发所有操作同时进行异步是各过各的。
3). 流量削峰商品秒杀 例如商品秒杀的时候这时候数据库并不能承受这么大的请求。可以把请求下订单的信息暂存在mq中返回给用户下单成功之后的操作由mq发送给对应的服务处理。缓存数据减少数据库的压力。 为什么需要使用消息中间件 服务解耦、异步通信、流量削峰 消息中间件在分布式系统中使用场景异步 6.1 服务解耦-订单和库存服务。用户下订单消息发给mqmq返回用户下订单成功消费者-库存服务接收mq消息再去调用减少库存的消息。 6.2 异步通信-用户注册新账户 用户注册和admin发送短信和邮件异步 6.3 流量削峰-商品秒杀先mq先存储订单信息返回订单服务下单成功后慢慢处理。减少大并发对数据库的影响/。 RabbitMQ的五种消息模型/工作模式、 1 simple 简单的一对一模式producce-queue-customer 2 word模式一个消息对列queue— 多个消费者消费者争抢消息队列里面消息注意一个消息只能被一个消费者消费。 3 fanout-广播、订阅者模式。交换机将消息发送给所有binding的对列消费端可以有多个customer使用word模式消费对列的消息。 4 topic-主体模式生产者的消息按照不同的路由规则模糊匹配给不同满足条件的消息对列消费者再去消费对列中消息 5routeKey,路由键exchange-type-direct按照不同的路由键发送到对应的queue中。 消息中间件是异步还是同步 异步各干各的互不影响。异步并不是并发-同时请求一个请求,而是互不影响个干各的没有约束和先后顺序。received生产者的messagesend消息到消费者。二者是异步解耦合互不影响。 mq的消息确认机制confirmMQ如何避免消息丢失 . 对于生产者端来说主要有两种确认机制 a. message到broker后mq立马确认confirm并返回消息告知生产者消息发送成功如果失败也告知生产者并重新发送。 b. message到MQ之后如果消息对列没有received成功(queue存储msg成功)会确认并返回消息接收失败到生产者 a b 保证了生产者端不会丢失消息。 2. 对于消费者来说。 a. 消费者接收到queue的消息后默认自动确认queue删除该message。 b. 消费者接收到msg后对数据进行逻辑处理如果直接confirm-queue直接删除msg处理数据过程中可能会宕机消息丢失。 ----设置为手动confirm确认收货数据处理完再收货成功queue再去删除msg。也可以对数据不满退回到queue重新入队也可以直接删除数据。 c. 接收失败告知queue不会删除数据MQ重新发送消息-这种操作很常见 这样避免数据在消费者端丢失
1、2两种方式避免了mq的消息丢失。 MQ重复消费 1如何造成重复消费 (1) 生产者端传输到MQ-queue消息对列接收成功,MQ因为网络问题没有ack-producer,导致生产者又发送了一次消息到MQ。queue-customer-这样msg就被消费了两次。 2消费者端MQ-queue消息对列消息传到customer。一种是消费者没有接收成功因为网络问题没有ack queuequeue重复发送这种不会造成msg重复消费。另一种是消费者消费成功但是因为不可控因素没有ack queue消息对列重复发送mgs-to-customer-重复消费。 2解决方法 对于幂等性消息查询消费者重复消费也没有关系。 对于非幂等性消息消费者重复消费就会有影响了。 方法消费者在消费消息之前获取msg唯一id到redis进行存储判断setnx(判断是已经存在并存储key-value)。
Boolen flagstringRadisTemplate.opsForValue.setAbsent(id,value) 1-flagtruekey不存在未被消费c正常消费msg 2-falsekey存在已经被消费两种可能-正在消费或者完成消费-忘记告知ack-queue了无论哪种情况都直接丢弃。
注意一个问题如果redis显示有消费记录-且消费者正在消费此时消费者执行业务宕机了redis分布式锁会成死锁-解决方法在IfAbsent方法加上过期时间和单位。
一句话就是消费之前缓存中有消费记录则丢弃消息不二次消费。 redis缓存中没有消费记录则重复存入缓存并消费设计锁过期时间。
以下是消息中间件MQ的相关代码和配置信息
使用MQ的步骤 1在pom文件中加上依赖amqp dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency2 配置文件配置rabbit服务器的对应信息spring.rabbitmq host、portusernameps等spring.rabbitmq.hostrabbitmq服务器地址信息
spring.rabbitmq.port端口号
spring.rabbitmq.username账户name
spring.rabbitmq.password密码
spring.rabbitmq.virtual-host/
#1. 生产者发送message mq收到消息就确认回复到生产者
spring.rabbitmq.publisher-confirmstr
#2. queue消息对列接收生产者的消息失败就确认返回消息到生产操者
spring.rabbitmq.publisher-returnstrue
#3. 消费者接收queue消息对列的消息之后手动确认
spring.rabbitmq.listener.simple.acknowledge-modemanual3 服务启动类上面加上注解EnableRabbit-开启MQ 在springboot启动类加上 EnableRabbit-开启MQ
4 业务使用消息中间件存储消息的时候 1 创建交换机注意有不同类型的交换机 direct-fanout-topic public void createExchange() {
// 1. 创建direct类型的exchange 交换机的名字-hello.java.exchangeDirectExchange directExchange new DirectExchange(hello.java.exchange, true, false);
// 2. 声明交换机amqpAdmin.declareExchange(directExchange);log.info(exchange创建成功1111, hello.java.exchange);}2创建消息队列queue
public void createQueue() {
// 1. 创建队列-queue 队列名称-hello-java-queueQueue queue new Queue(hello.java.queue, true, false, false);
// 2. 声明mq队列amqpAdmin.declareQueue(queue);log.info(queue创建成功1111, hello.java.queue);}
3交换机和消息队列直接关系绑定 public void bindEQ() {
// 1. 创建绑定对象 hello.java.queue--消息对列 hello.java.exchange--交换机hello.java-绑定关系的route-keyBinding binding new Binding(hello.java.queue,Binding.DestinationType.QUEUE,hello.java.exchange,hello.java,null);// 2. 声明绑定关系这个关系实际也是一个对象amqpAdmin.declareBinding(binding);log.info(Binding创建成功1111, hello.java.binding);}4使用MQ的操作工具类 RabbitTemplate-操作发送消息 对象注入 AutowiredRabbitTemplate rabbitTemplate;生产者发送消息需要携带消息-mgs和发送给哪个queue的route-key。注意发送消息需要一个唯一id后面防止重复发送需要此id判断 public void sendMessageStr() throws InterruptedException {String msg 测试数据测试数;
// 发送10条message到exchange中
// new CorrelationData(UUID.randomUUID().toString() 发送的消息的唯一id mq可以接收并处理rabbitTemplate.convertAndSend(hello.java.exchange, hello.java , msg 11111111111111, new CorrelationData(UUID.randomUUID().toString()));rabbitTemplate.convertAndSend(hello.java.exchange, hello.java, msg 222222222222, new CorrelationData(UUID.randomUUID().toString()));log.info(交换机消息发送成功-----------);}4消费者监听消息对列消息消费消息使用RabbitListener监听消息对列使用RabbitHandler接收对应类型的消息。前者放在类上面后者放到监听方法上面。queues是消息对列名称的集合
RabbitListener(queues {hello.java.queue})使用RabbitHandler监听不同类型的消息
// 消息是TestEntity2 类型会自动匹配到对应方法接收
RabbitHandler
public void receiveOfSecond(TestEntity2 testEntity2) throws InterruptedException {System.out.println(receiveOfSecond-监听接受queue的数据是----- testEntity2);
}RabbitHandler
public void receiveOfFirst(TestEntity testEntity) throws InterruptedException {System.out.println(receiveOfFirst-监听接受queue的数据是----- testEntity);
}
