任务调度框架-如何实现定时任务+RabbitMQ事务+手动ACK

任务调度框架

Java中如何实现定时任务？

比如：
1.每天早上6点定时执行
2.每月最后一个工作日，考勤统计
3.每个月25号信用卡还款
4.会员生日祝福
5.每隔3秒，自动提醒

10分钟的超时订单的自动取消，每隔30秒或1分钟查询一次订单，拿当前的时间上前推10分钟
定时任务，资源会有误差的存在，如果使用定时任务
定时任务，用于统计的时候最多。

自动统计考勤，一般0点之后开始统计，可以使用定时任务

nacos心跳

晚上要求和采购部门生成采购单，达到最低预警值的时候，去发给采购部门

我们可以通过任务调度框架实现上述的需求
任务调度框架，可以实现定时任务，实现间隔多少时间的重复执行，实现指定日期的重复执行

电商自动好评，间隔时间长的，误差几分钟，影响不大。

java中任务调度框架：
1、Spring Task spring自带的
2、Quartz 古老的框架
3、XXL-Job
4、第三方云平台-比如说：阿里云-SchedulerX等等
选择一个:Spring Task
2个注解+
包：task任务、job

使用步骤：
1、开关类，使用注解
@EnableScheduling // 开启任务调度
在这里插入图片描述
2、定义定时任务

@Scheduled(cron = "0/3 * * * ?")

CORN表达式：
秒 分 时 日 月 星期几 年 其中，只有年可以省略
定时任务，需要重复执行的方法

CORN表达式：
特殊字符串，主要用来描述时间的，用于任务调度等
https://cron.qqe2.com/

/ 间隔
- 是连续
, 枚举值
L 最后，星期、日中用
W 有效工作日
LW 某个月最后一个工作日
# 用于确定每个月第几个星期几，母亲节或父亲节
4#2 某个月的第二个星期三  4代表星期三，中文的时候有可能不影响

在这里插入图片描述

在这里插入图片描述

项目名：SpringTask01
Spring Web、Lombok

RabbitMq实现延迟：

死信+延迟消息处理

死信：RabbitMQ的队列中的消息，满足以下条件任意其一就会成为死信消息：
1.消息被拒绝
2.消息过期
3.队列满了
死信交换器:专门用来转发队列中的死信消息，将死信消息转发到指定的队列中

十分钟未支付，取消订单？
十分钟之后，消息会过期，过期后，通过死信交换器转发队列中的死信消息，将死信消息转发到指定的队列中，由消费者去处理。

我们可以通过死信+死信交换器实现延迟消息处理
RabbitMQ实现延迟消息处理有2种方式：
1、死信+死信交换器 代码实现（可控，更方便一些）
2、延迟消息插件

1、死信+死信交换器 代码实现（可控，更方便一些）
发送消息，到队列1，（产生延迟队列，产生死信）
在这里插入图片描述
一个消息，过一段时间，实现消费。

核心：
1.队列 是2个队列，第1个队列： 目的 产生死信（1.设置有效期2.不设置消费者），借助死信交换器，把产生的死信发到指定的队列中
第二个对了：目的 消费死信，这里获取的信息，时间就是延迟的，延迟的就是上面的有效期
2.1个交换器
死信交换器，整个系统一般就一个，可以用来转发各个功能产生的死信，是Direct类型的交换，通过RK进行消息匹配到对应的队列中。
RabbitMQ的消息的有效期有2种设置方式：
1.设置队列上的有效期，整个队列中所有消息都使用
2.可以在每个消息上设置有效期，这种适用于有多个不同有效期的消息

在这里插入图片描述
如果队列和消息都有有效期，谁短听谁的。

代码：
RabbitMQ02
实现：
1.pom

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><!--        RabbitMQ --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency></dependencies>

2.代码：
config-RabbitMQConfig

package com.yd.rabbitmq02.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;/*** @author MaoXiqi* @organization: Lucky* @create 2023-10-16 11:35:54* @description RabbitMQConfigApi*/
@Configuration
public class RabbitMQConfig {// 1.创建2个队列@Beanpublic Queue createQ1() {// 1.设置队列 内部消息有效期 设置死信交换器 设置RKHashMap<String, Object> params = new HashMap<>();// 设置队列中每个消息的有效期 单位 毫秒params.put("x-message-ttl", 3000);// 设置对应的死信交换器params.put("x-dead-letter-exchange", "dead-ex-yd");// 设置交换器匹配的路由名称params.put("x-dead-letter-routing-key", "test");return QueueBuilder.durable("dl-q01").withArguments(params).build();}@Beanpublic Queue createQ2() {return new Queue("dl-q02");}// 2.创建1个交换器（1.fanout 2，direct 3.topic 4.header）-死信交换器direct类型@Beanpublic DirectExchange createDe() {return new DirectExchange("dead-ex-yd");}// 3.创建1个绑定@Beanpublic Binding createBd1(DirectExchange de) {return BindingBuilder.bind(createQ2()).to(de).with("test");}
}

在这里插入图片描述

2.1.创建两个队列
1、设置队列，内部消息有效期 设置死信交换器 设置RK 注意：参数名是固定的，值根据业务需求去改，值 单位是毫秒

2.2创建1个交换器（1.fanout 2.direct 3.tipic
4.header）-死信交换器direct类型

2.3.创建一个绑定

controller-DeadController

    @GetMapping("send")public String sendDead(String msg) {System.out.println("发送消息," + msg + ",发送时间:" + System.currentTimeMillis());template.convertAndSend("", "dl-q01", msg);return "ok";}

发送
在这里插入图片描述
监听消息-主要是为了消费：
listener-DeadListener

    @RabbitListener(queues = "dl-q02")public void handler(String m) {System.out.println("延迟消息,"+m+",接受时间:" +System.currentTimeMillis());}

在这里插入图片描述
yml：

spring:rabbitmq:host: 121.36.5.100port: 5672username: guestpassword: guest
server:port: 8082

测试：
在这里插入图片描述

RabbitMQ事务：

数据库中事务：保证数据一致性，特别是多个操作要么都成功，要么都失败
RabbitMQ事务：一次性发多条消息，需要开启事务，
RabbitMQ也有自己的事务，如果本次在这里插入图片描述

在这里插入图片描述

使用步骤：

源码：RabbitMQ02
1.创建配置类
2.使用基于事务发送
3.监听消息

1.创建配置类
config-RabbitMQTranConfig
1.准备一个队列
2.创建事务管理器

// 创建事务管理器@Beanpublic RabbitTransactionManager createTran(ConnectionFactory factory) {return new RabbitTransactionManager(factory);}

在这里插入图片描述

2.controller-TranController
开启事务
发送消息

    // 事务@Transactional // 需要开启SpringBoot的事务机制@GetMapping("sendmsg")public String sendMsg(String msg, int count) {// 开启 RabbitMQ的通道的事务template.setChannelTransacted(true);// 发送消息for (int i = 0; i < count; i++) {template.convertAndSend("", "yd-tran-q01", msg + "--" + count);// 出错，看看 事务是否生效if (i==2) {System.out.println(1/0);}}return "ok";}

在这里插入图片描述

3.listener-DeadListener

    // 事务@RabbitListener(queues = "yd-tran-q01")public void handler2(String msg) {System.out.println("监听消息"+msg);// 处理业务逻辑 出错了}

在这里插入图片描述

在这里插入图片描述

手动ACK

RabbitMQ怎么防止消息丢失:
1.发送端没有发送过去
解决：
1.用事务
2.confirm消息确认机制 万能：转人工处理
2.MQ服务器丢失，MQ服务器蹦了，
解决：开启持久化
3.消费端消息丢失：
解决：自动应答，改成开始手动ACK

消息确认机制：默认是自动确认

消息的发送和接收是异步

RabbitMQ如何防止消息丢失：
1.
代码：
config-RabbitMqTranConfig

    //消费消息的⼿动应答@Beanpublic Queue createQ4() {return new Queue("yd-ack-q01");}

controller-DeadController

    // 事务@Transactional // 需要开启SpringBoot的事务机制@GetMapping("sendmsg")public String sendMsg(String msg, int count) {// 开启 RabbitMQ的通道的事务template.setChannelTransacted(true);// 发送消息for (int i = 0; i < count; i++) {template.convertAndSend("", "yd-tran-q01", msg + "--" + count);// 出错，看看 事务是否生效if (i==2) {System.out.println(1/0);}}return "ok";}

listener-DeadListener
一般设置个上限，比如最多三次

    @RabbitListener(queues = "yd-ack-q01")public void handler3(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {//消费者获取消息，默认采用的自动应答，就是获取就应答，这样MQ服务器就删除消息//还可以手动应答:结果:1.成功(MQ删除)2.失败(MQ消息)System.out.println("收到ACK消息，监听消息:" + msg);//拒绝消息 参数说明：1.消息id 2.结果 true 成功 false 拒绝 3.是否把消息添加回队列中channel.basicNack(tag,false,true);//成功消息 参数说明：1.消息id 2.结果 true 成功 false 拒绝//channel.basicAck(tag,true);// 处理业务逻辑 出错了}

在这里插入图片描述
在这里插入图片描述

消费消息的手动应答：
RabbitMQ默认的消费者消息获取模式采用的是手动应答
但是这种有缺陷，可能会出现，消息获取了但是业务出了问题，导致MQ也自动删除了消息，最终导致业务没有执行
所以为了解决这种问题，可以开启手动应答模式，结合自己的业务执行情况，如果业务执行成功，那么就成功应答，如果失败了，就拒绝消息，同时把消息再加回队列，这样就可以再次消息再次处理（加个上限）
在这里插入图片描述
测试
在这里插入图片描述

RabbitMQ如何保证消息的幂等性：
幂等性就是重复消费。
解决：
1.生成一个全局id，存入redis或者数据库，在消费者消费消息之前，查询一下该消息是否有小费过。
2.
在这里插入图片描述

用户充值，重复消费，相当如充值了多次，是一定要杜绝的。

在这里插入图片描述

不要返回值的时候，可以用RabbitMQ替代OpenFegin，因为消费完就不回了。MQ默认是单向的
短信发信可以用MQ，这个业务要做，做起来可能会很耗时，中间要经过运营商，过程不可控

RabbitMQ应用场景
消息通信，发送消息和接收消息，是异步
1.实现服务通信
用在微服务中，实现2个服务的通信，这种不带返回值的，只是为了执行另一个服务的方法执行
2.解决耗时操作
比如：邮件、短信、第三方接口等，比较耗时
3.解耦
4.提升性能
5.重复代码封装
6.削峰填谷（订单先下到redis中，再通过MQ和延迟队列，慢慢的从redis搬到mysql中）

关键词：异步、解耦、延迟

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/139263.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

CentOS有IP地址，连接不上Xshell或使用Xshell时突然断开

CentOS有IP地址，连接不上Xshell或使用Xshell时突然断开

问题原因&#xff1a;未在电脑主机的网络中进行IP地址配置 解决办法&#xff1a; 1.打开控制面板&#xff0c;选择‘网络与共享中心’ 2.选择“更改适配器设置” 3.右键点击以太网3“属性” 4.选择协议版本4&#xff0c;点击属性 5.IP地址填写CentOS的IP地址&#xff1a;192.…
阅读更多...
搭建伪分布式Hadoop

搭建伪分布式Hadoop

文章目录 一、Hadoop部署模式&#xff08;一&#xff09;独立模式&#xff08;二&#xff09;伪分布式模式&#xff08;三&#xff09;完全分布式模式 二、搭建伪分布式Hadoop&#xff08;一&#xff09;登录虚拟机&#xff08;二&#xff09;上传安装包&#xff08;三&#xf…
阅读更多...
ios UIDocumentPickerViewController 实现TEXT、DOC、PDF等文档读取

ios UIDocumentPickerViewController 实现TEXT、DOC、PDF等文档读取

文章目录 一、前言二、iCould相关配置三、功能实现3.1 UIDocumentPickerViewController 选取控制器3.2 读取文件一、前言 最近正在研发的项目有一个需求： 允许用户将iCloud中的文档上传，实现文件的流转。 以前接触的项目对于资料类的上传大多是仅限于图片与视频。对于文档类…
阅读更多...
干洗店小程序，上门洗鞋洗衣下单算软件开发；

干洗店小程序，上门洗鞋洗衣下单算软件开发；

干洗店小程序&#xff0c;上门洗鞋洗衣软件&#xff1b; 有多个小程序开发案例 洗衣店小程序 是一款适用于洗衣店、干洗店等洗鞋店小程序&#xff1b; 干洗店小程序功能有: 1.(支持上门取送、送货到店、寄存网点、智能衣柜四种下单方式) 用户下单-上门取货拍照-送达门店工厂-入…
阅读更多...
【快刊推荐】综合类EI，最新案例29天录用，国人友好！

【快刊推荐】综合类EI，最新案例29天录用，国人友好！

综合类 • 快刊推荐 01 期刊简介 检索数据库&#xff1a;EI &Scopus &Google Scholar 版面类别&#xff1a;正刊 数据库收录年份&#xff1a;2009年 国人占比&#xff1a;47%&#xff0c;对国人友好 年发文量&#xff1a;20篇左右 02 接收领域 生物&#xff…
阅读更多...
云服务器ip使用细节（公网、私有）

云服务器ip使用细节（公网、私有）

场景&#xff1a; 当我们对tcp服务器进行监听的时候&#xff0c;可能需要用到ip地址&#xff0c;比如使用httplib::Service::listen(ip, port)&#xff0c;而当我们访问tcp服务器时也需要ip地址 但这两个ip是不同的&#xff01; 每个云服务器通常都会有一个公网IP地址和一个私有…
阅读更多...
哈佛教授因果推断力作：《Causal Inference: What If 》pdf下载

哈佛教授因果推断力作：《Causal Inference: What If 》pdf下载

因果推断是一项复杂的科学任务&#xff0c;它依赖于多个来源的三角互证和各种方法论方法的应用&#xff0c;是用于解释分析的强大建模工具&#xff0c;同时也是机器学习领域的热门研究方向之一。 今天我要给大家推荐的这本书&#xff0c;正是因果推断领域必读的入门秘籍&#…
阅读更多...
深入理解强化学习——智能体的类型：有模型强化学习智能体与免模型强化学习智能体

深入理解强化学习——智能体的类型：有模型强化学习智能体与免模型强化学习智能体

分类目录&#xff1a;《深入理解强化学习》总目录 根据智能体学习的事物不同&#xff0c;我们可以把智能体进行归类。基于价值的智能体&#xff08;Value-based agent&#xff09;显式地学习价值函数&#xff0c;隐式地学习它的策略。策略是其从学到的价值函数里面推算出来的。…
阅读更多...
基础课4——语音识别技术

基础课4——语音识别技术

ASR 是自动语音识别&#xff08;Automatic Speech Recognition&#xff09;的缩写&#xff0c;是一种将人类语音转换为文本的技术。ASR 系统可以处理实时音频流或已录制的音频文件&#xff0c;并将其转换为文本。它是一种自然语言处理技术&#xff0c;广泛应用于许多领域&#…
阅读更多...
如何通过SK集成chatGPT实现DotNet项目工程化？

如何通过SK集成chatGPT实现DotNet项目工程化？

智能助手服务 以下案例将讲解如何实现天气插件 当前文档对应src/assistant/Chat.SemanticServer项目 首先我们介绍一下Chat.SemanticServer的技术架构 SemanticKernel 是什么&#xff1f; Semantic Kernel是一个SDK&#xff0c;它将OpenAI、Azure OpenAI和Hugging Face等大…
阅读更多...
React Hooks之useReducer

React Hooks之useReducer

一、useReducer const [state, dispatch] useReducer(reducer, initialArg, init?)useState 的替代方案。它接收一个形如 (state, action) > newState 的 reducer&#xff0c;并返回当前的 state 以及与其配套的 dispatch 方法。数据结构简单用useState&#xff0c;复杂时…
阅读更多...
linux虚机新增加磁盘后在系统中查不到

linux虚机新增加磁盘后在系统中查不到

问题描述 在虚机管理平台上对某一linux主机添加了一块硬盘&#xff0c;但在系统中并未显示 通过执行 lsblk&#xff0c;并未看到新增的硬盘信息 解决方法 1. 可通过重启服务器解决 2. 如果不能重启服务器&#xff0c;可重新扫描下 scsi总线 查看总线&#xff1a; ls /s…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023