ddddd

项目二阶段总结

账户微服务

短信发送

1.压测发现问题

首先对短信smscomponent的send方法在test单元测试类中测试，不是真的发短信测试，可以建立请求开始和结束的时间戳来确定请求的耗时。

1.在notifycontroller用notify serviceimpl中实现notify service中的test方法测试，

//        ResponseEntity<String> forEntity = restTemplate.getForEntity("http://old.xdclass.net", String.class);

//        String body = forEntity.getBody();

//        long endTime = CommonUtil.getCurrentTimestamp();

//        log.info("耗时={},body={}",endTime-beginTime,body);

RestTemplate 发送的是 HTTP 请求，那么在响应的数据中必然也有响应头，如果开发者需要获取响应头的话，那么就需要使用 getForEntity 来发送 HTTP 请求，此时返回的对象是一个 ResponseEntity 的实例。这个实例中包含了响应数据以及响应头。

getForEntity 方法。第一个参数是 url ，url 中有一个占位符 {1} ,如果有多个占位符分别用 {2} 、 {3} … 去表示，第二个参数是接口返回的数据类型，最后是一个可变长度的参数，用来给占位符填值。在返回的 ResponseEntity 中，可以获取响应头中的信息，其中 getStatusCode 方法用来获取响应状态码， getBody 方法用来获取响应数据， getHeaders 方法用来获取响应头。

目前用的是同步发送+resttemplate未池化，压测的是短信发送服务，用户请求到短信服务，然后http请求调用第三方短信服务，响应完之后返回给用户通知，很慢。400-500.

2.加Async注解实现异步

启动类里面使用@EnableAsync注解开启功能，自动扫描或者创建一个配置类（ThreadPoolTaskConfig）加上@EnableAsync
定义异步任务类并使用@Component标记组件被容器扫描,异步方法加上@Async
注意Async失效，不能调用public和ststic方法以及，必须在外部的类中调用这个方法

3.自定义线程池和Async注解配合

核心思想：使用Async注解创建线程任务，并且指定自定义的线程池进行任务的消费

@Async 注解中有一个 value 属性，看注释应该是可以指定是哪个线程池的

@Async使用的线程池Bean名称为applicationTaskExecutor，显而易见这个队列设置为Integer.MAX_VALUE，所以在上面会导致堆内存溢出

于是要自定义线程池，线程池顺序先是CorePoolSize是否满足，然后是Queue阻塞队列是否满，最后才是MaxPoolSize是否满足

corePoolSize：核心线程数

maximumPoolSize：最大线程数

maximumPoolSize - corePoolSize = 救急线程数

keepAliveTime：救急线程空闲时的最大生存时间

unit：时间单位

workQueue：阻塞队列（存放任务）

有界阻塞队列 ArrayBlockingQueue

无界阻塞队列 LinkedBlockingQueue

最多只有一个同步元素的队列 SynchronousQueue

优先队列 PriorityBlockingQueue

threadFactory：线程工厂（给线程取名字）

handler：拒绝策略

工作方式：

线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。

当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入 workQueue 队列排 队，直到有空闲的线程。

如果队列选择了有界队列，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线 程来救急。

如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 下面的前 4 种实现，其它著名框架也提供了实现

ThreadPoolExecutor.AbortPolicy 让调用者抛出RejectedExecutionException 异常，这是默认策略

ThreadPoolExecutor.CallerRunsPolicy 让调用者运行任务

ThreadPoolExecutor.DiscardPolicy 放弃本次任务

ThreadPoolExecutor.DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之

Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方 便定位问题

Netty 的实现，是创建一个新线程来执行任务

ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略

PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略

当高峰过去后，超过 corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由 keepAliveTime 和 unit 来控制。

https://blog.csdn.net/weixin_45325628/article/details/122125768 【多线程】@Async注解和线程池

https://wenku.baidu.com/view/aeacea08b7daa58da0116c175f0e7cd1842518e7.html Springboot注解@Async线程池实例详解

对比和问题

异步发送 + resttemplate未池化

线程池参数

threadPoolTaskExecutor.setCorePoolSize(4);

threadPoolTaskExecutor.setMaxPoolSize(16);

threadPoolTaskExecutor.setQueueCapacity(32);

qps少，等待队列小

异步发送+resttemplate未池化

线程池参数

threadPoolTaskExecutor.setCorePoolSize(32);

threadPoolTaskExecutor.setMaxPoolSize(64);

threadPoolTaskExecutor.setQueueCapacity(10000);

//如果等待队列长度为10万，则qps瞬间很高8k+,可能oom

qps，等待队列大(瞬间高)

问题
- 采用异步发送用户体验变好了，但是存在丢失的可能，阻塞队列存储内存中，如果队列长度过多则重启容易出现丢失数据情况
- 采用了异步发送了+阻塞队列存缓冲，刚开始瞬间QPS高，但是后续也降低很多
- 问题是在哪里？消费方角度，提高消费能力

4.异步线程池加http连接池

首先要自定义线程池，海量请求下可能阻塞队列里可能无线增长，或者oom。还要提高消费性能，用http连接池

从消费方角度，阻塞队列里的任务是向第三方http请求，如果能让这个HTTP请求消费的快一点，那么阻塞队列中的任务存储的数量就会变少。要提升restTemplate性能。

我们可以看到串行连接时，客户端每一次发送请求便会打开一个连接，然后响应结束之后关闭连接。这种情况下，连接的建立和销毁都会消耗很多额外资源。而持久连接，则是第一次请求到达时会打开连接，随后连接不会被关闭，等到下一次请求到达。

同线程池概念差不多，连接池也是采用了“池”的概念达到了复用：

当有连接第一次使用的时候建立连接
结束时对应连接不关闭，归还到池中
下次同个目的的连接可从池中获取一个可用连接
定期清理过期连接

restTemplate底层实现httpClient，封装了关于http连接池的一系列方法，而我们只需要通过配置文件的形式设置超时时间，最大连接数、每路由的最大连接数等等即可。而达到复用是通过连接的route（IP+PORT）从每个route的池中获得连接，如若超过数量，则创建或等到，知道超时。

在coommon包中的restTemplateconfig包中进行设置。

springboot集成restTemplate，http连接池 https://blog.csdn.net/qq_32302897/article/details/84550377

http结构

https://blog.csdn.net/weixin_45598506/article/details/112917752

GET /tutorials/other/top-20-mysql-best-practices/ HTTP/1.1

Host: net.tutsplus.com User-Agent: Mozilla/5.0 (Windows; U; Windows NT 6.1; en-US; rv:1.9.1.5) Gecko/20091102 Firefox/3.5.5 (.NET CLR 3.5.30729)

Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8Accept-Language: en-us,en;q=0.5

Accept-Encoding: gzip,deflate

Accept-Charset: ISO-8859-1,utf-8;q=0.7,*;q=0.7

Keep-Alive: 300

Connection: keep-alive

Cookie: PHPSESSID=r2t5uvjq435r4q7ib3vtdjq120Pragma: no-cacheCache-Control: no-cache

--------------------------------------------------------------------------------------

HTTP/1.x 200 OK

Transfer-Encoding: chunkedDate: Sat, 28 Nov 2009 04:36:25

GMTServer: LiteSpeedConnection: closeX-Powered-By: W3 Total Cache/0.8Pragma: publicExpires: Sat, 28 Nov 2009 05:36:25 GMTEtag: "pub1259380237;gz"Cache-Control: max-age=3600, public

Content-Type: text/html; charset=UTF-8

Last-Modified: Sat, 28 Nov 2009 03:50:37

 GMTX-Pingback: http://net.tutsplus.com/xmlrpc.php

Content-Encoding: gzipVary:

 Accept-Encoding,

 Cookie, User-Agent<!-- ... rest of the html ...

第一行是start line（起始行），request line 或者stautus line，描述请求或者响应的基本信息

接下来就是header，以key value形式来详细说明报文

消息正文，entity，实际传输的数据，不一定是纯文本，可以是音乐，视频等二进制的数据

短信验证码总结

https://zhuanlan.zhihu.com/p/373732960 异步http

@Async("threadPoolTaskExecutor")

public void send(String to,String templateId,String value){

//to手机号,模板id，value

    String url = String.format(URL_TEMPLATE,to,templateId,value);//构建发送的url， to是手机号

    HttpHeaders headers = new HttpHeaders();

    //最后在header中的格式(中间是英文空格)为Authorization:APPCODE 83359fd73fe94948385f570e3c139105

    headers.set("Authorization","APPCODE "+smsConfig.getAppCode());//appcode后面有个空格

    HttpEntity entity = new HttpEntity<>(headers);

    ResponseEntity<String> response = restTemplate.exchange(url, HttpMethod.POST, entity, String.class);

    log.info("url={},body={}",url,response.getBody());

    if(response.getStatusCode().is2xxSuccessful()){

        log.info("发送短信验证码成功");

    }else {

        log.error("发送短信验证码失败:{}",response.getBody());

    }

}

1.短信发送的时候，参考短信服务的商家的文档，里面规定了发送短信的方法规范，方法有三个参数，分别是要发送的手机号，还有模板id，这个模板id由服务商提供，还有一个参数value，存储短信的内容。首先要指定http的请求头，设定相应的keyvalue，用作鉴权，key就是Authorization字符串，value是商户号。然后通过restTemplate对商家指定的url进行post请求来得到相应的响应，如果响应码是2开头就是请求成功。

2.然后接下来对这个过程进行优化，首先这个过程是用户请求短信服务，短信服务通过http请求第三方服务商，然后第三方服务返回响应，然后短信服务告知用户结果，这个是一个同步发送的过程。第一个请求未完成，第二个请求也发不出去，所以采用异步发送的方式，

通过采用@Async注解和自定义线程池的方式来("threadPoolTaskExecutor")实现，采用自定义线程池能自定义线程池的参数，比如maxpoolsize，默认的线程池的maxpoolsize的大小是Integer.max，所以当任务很多的时候，会引起堆内存溢出，或者当阻塞队列中任务很多的时候，这时候如果宕机，就会引起消息丢失，所以自定义线程池中要自定义最大线程数的大小。

3.最后，海量请求下可能阻塞队列里可能无线增长，或者oom。还要提高消费性能，用http连接池

从消费方角度看，阻塞队列里的任务是向第三方http请求，而且每次http请求都要三次握手，四次挥手来开启和关闭，如果能让这个HTTP请求消费的快一点，那么阻塞队列中的任务存储的数量就会变少。所以就采用了http连接池，http连接池也是和线程池一样的池化思想。

1.当有连接第一次使用的时候建立连接

2.结束时对应连接不关闭，归还到池中

3.下次同个目的的连接可从池中获取一个可用连接

4.定期清理过期连接

然后实现http连接池，其实就是通过对restTemplate方法改写，因为他的底层实现httpClient，封装了关于http连接池的一系列方法，而我们只需要通过配置文件的形式设置超时时间，最大连接数、每路由的最大连接数等等即可。而达到复用是通过连接的route（IP+PORT）从每个route的池中获得连接，如若超过数量，则创建或等到，知道超时。

4.最后发送短信验证码就通过异步线程池加上http连接池的方法提高了QPS，我用jmter自己压测的时候，在相同配置下，从最开始采用同步发送和restTemplte未采用http连接池，此时的qps只有400-500左右，最后提高了10倍到了3000-4000左右。而且我自己压测机器和开发运行服务的机器是在同一台电脑上，会有性能损耗，实际的QPS会更高。

验证码

1.redis连接池

连接池好处

使用连接池不用每次都走三次握手、每次都关闭Jedis
相对于直连，使用相对麻烦，在资源管理上需要很多参数来保证，规划不合理也会出现问题
如果pool已经分配了maxActive个jedis实例，则此时pool的状态就成exhausted了

redis整合Spring时为何需要在存储的时候序列化？https://www.zhihu.com/question/277363840/answer/392945240

https://www.cnblogs.com/hzhl/articles/13974129.html redis的序列化和反序列化

任何存储都需要序列化。只不过常规你在用DB一类存储的时候，这个事情DB帮你在内部搞定了（直接把SQL带有类型的数据转换成内部序列化的格式，存储；读取时再解析出来）。而Redis并不会帮你做这个事情。当你用Redis的key和value时，value对于redis来讲就是个byte array。你要自己负责把你的数据结构转换成byte array，等读取时再读出来。

不重写序列化规则可能会乱码

Redis连接池技术

https://blog.csdn.net/csdn_shilin/article/details/109147790

https://blog.csdn.net/yaomingyang/article/details/79036439?utm_medium=distribute.pc_relevant.none-task-blog-2_defaultbaidujs_baidulandingword~default-1.pc_relevant_default&spm=1001.2101.3001.4242.2&utm_relevant_index=4

https://blog.csdn.net/lihongfei110/article/details/106729300/?utm_medium=distribute.pc_relevant.none-task-blog-2_defaultbaidujs_baidulandingword~default-1.pc_relevant_default&spm=1001.2101.3001.4242.2&utm_relevant_index=4

redis和jedis

https://zhuanlan.zhihu.com/p/134682772

https://blog.csdn.net/qq_40325734/article/details/81072665

https://blog.csdn.net/zxl646801924/article/details/82770026

用户注册服务

首先获取四位验证码，用谷歌kaptcha框架实现生成，并且将验证码存入redis中，设置相应的过期时间。这里用到了String类型，存入的key是user-service:captcha+ip和http请求头的user—agent的md5加密字符串，value是验证码。在获取手机验证码的时候，先通过key找到redis中的验证码value，如果输入的验证码正确，那么开始发送手机的验证码，并且在redis中删除相应的验证码数据。

注册手机验证码做了的60秒内防重提交，把验证码拼接上时间戳，比如161111_20222222,并且与用户输入相匹配，通过字符串的拼接来得到之前的时间戳，并且通过当前时间戳与之前时间戳相比较，如果在60内，并且验证码输入正确就可以注册。

在这注册过程中要设置密码，密码的存储为盐加密码再进行md5加密然后存入数据库中

通过手机号进行账号唯一性检查，因为数据库的手机号是唯一索引。

调用优惠卷微服务的领券接口来对新人发放优惠券。

短链(link)微服务

分库分表

数据库性能优化思路讲解（需要补充详细）

- 千万不要一上来就说分库分表，这个是最忌讳的事项
- 一定要根据实际情况分析，两个角度思考
  - 不分库分表
    - 软优化
      - 数据库参数调优
      - 分析慢查询SQL语句，分析执行计划，进行sql改写和程序改写
      - 优化数据库索引结构
      - 优化数据表结构优化
      - 引入NOSQL和程序架构调整：大多数业务都是读多写少，所以主从分离，让从库执行读操作，在主库执行写操作
    - 硬优化
      - 提升系统硬件（更快的IO、更多的内存）：带宽、CPU、硬盘
  - 分库分表
    - 根据业务情况而定，选择合适的分库分表策略（没有通用的策略）
      - 外卖、物流、电商领域
    - 先看只分表是否满足业务的需求和未来增长
      - 数据库分表能够解决单表数据量很大的时,数据查询的效率问题，
      - 无法给数据库的并发操作带来效率上的提高，分表的实质还是在一个数据库上进行的操作，受数据库IO性能的限制
    - 如果单分表满足不了需求，再分库分表一起
结论
- 在数据量及访问压力不是特别大的情况，首先考虑缓存、读写分离、索引技术等方案
- 如果数据量极大，且业务持续增长快，再考虑分库分表方案

垂直分库

垂直分库是指按照业务将表进行分类，分布到不同的数据库上面，每个库可以放在不同的服务器上，它的核心理念

是专库专用。

它带来的提升是：1.解决业务层面的耦合，业务清晰

2.能对不同业务的数据进行分级管理、维护、监控、扩展等

3.高并发场景下，垂直分库一定程度的提升IO、数据库连接数、降低单机硬件资源的瓶颈

垂直分库通过将表按业务分类，然后分布在不同数据库，并且可以将这些数据库部署在不同服务器上，从而达到多

个服务器共同分摊压力的效果，但是依然没有解决单表数据的过大的问题。

水平分库

水平分库是把同一个表的数据按一定规则拆到不同的数据库中，每个库可以放在不同的服务器上。

它带来的提升是：解决了单库大数据，高并发的性能瓶颈。提高了系统的稳定性及可用性。

当一个应用难以再细粒度的垂直切分，或切分后数据量行数巨大，存在单库读写、存储性能瓶颈，这时候就需要进

行水平分库了，经过水平切分的优化，往往能解决单库存储量及性能瓶颈。但由于同一个表被分配在不同的数据

库，需要额外进行数据操作的路由工作，因此大大提升了系统复杂度。

垂直分表

垂直分表定义：将一个表按照字段分成多表，每个表存储其中一部分字段。它带来的提升是：

1.为了避免IO争抢并减少锁表的几率，查看详情的用户与商品信息浏览互不影响

2.充分发挥热门数据的操作效率，商品信息的操作的高效率不会被商品描述的低效率所拖累。

一般来说，某业务实体中的各个数据项的访问频次是不一样的，部分数据项可能是占用存储空间比较大的BLOB或是TEXT。例如上例中的商品描述。所以，当表数据量很大时，可以将表按字段切开，将热门字段、冷门字段分开放置在不同库中，这些库可以放在不同的存储设备上，避免IO争抢。垂直切分带来的性能提升主要集中在热门数据的操作效率上，而且磁盘争用情况减少。

通常我们按以下原则进行垂直拆分:

把不常用的字段单独放在一张表;
把text，blob等大字段拆分出来放在附表中;
经常组合查询的列放在一张表中;

水平分表

水平分表是在同一个数据库内，把同一个表的数据按一定规则拆到多个表中。它带来的提升是：优化单一表数据量过大而产生的性能问题避免IO争抢并减少锁表的几率库内的水平分表，解决了单一表数据量过大的问题，分出来的小表中只包含一部分数据，从而使得单个表的数据量变小，提高检索性能。

traffic水平分表

CREATE TABLE `traffic` (

  `id` bigint unsigned NOT NULL AUTO_INCREMENT,

  `day_limit` int DEFAULT NULL COMMENT '每天限制多少条，短链',

  `day_used` int DEFAULT NULL COMMENT '当天用了多少条，短链',

  `total_limit` int DEFAULT NULL COMMENT '总次数，活码才用',

  `account_no` bigint DEFAULT NULL COMMENT '账号',

  `out_trade_no` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '订单号',

  `level` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '产品层级：FIRST青铜、SECOND黄金、THIRD钻石',

  `expired_date` date DEFAULT NULL COMMENT '过期日期',

  `plugin_type` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '插件类型',

  `product_id` bigint DEFAULT NULL COMMENT '商品主键',

  `gmt_create` datetime DEFAULT CURRENT_TIMESTAMP,

  `gmt_modified` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

  PRIMARY KEY (`id`),

  UNIQUE KEY `uk_trade_no` (`out_trade_no`,`account_no`) USING BTREE,

  KEY `idx_account_no` (`account_no`) USING BTREE

) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

比较重要的是 out_trade_no account_no 设为唯一索引

product_id 是商品主键

分表数量：线上分8张表，本地分2张表即可
分片key： account_no，查询维度都是根据account_no进行查询
分片策略：行表达式分片策略 InlineShardingStrategy

步骤：1.在account数据库中新建两个traffic0和traffic1的表2.在idea中配置相应的环境（pom文件中加入 sharding-jdbc依赖包）3.在application.properties里面添加配置和分片策略，根据account_no%2分表，分别进入traffic_0和traffic_1表

      table-strategy:

        inline:

          algorithm-expression: traffic_$->{account_no % 2}

          sharding-column: account_no

引起的问题：traffic表中的主键id重复，单库下一般使用Mysql自增ID, 但是分库分表后，会造成不同分片上的数据表主键会重复。需求1.性能强劲2.全局唯一3.防止恶意用户根据id的规则来获取数据

采用雪花算法生成id

雪花算法生成的数字，long类，所以就是8个byte，64bit

表示的值 -9223372036854775808（-2的63次方） ~ 9223372036854775807（2的63次方-1）
生成的唯一值用于数据库主键，不能是负数，所以值为0~9223372036854775807（2的63次方-1）

64bit

 1+41+5+5+12

 1.第一位bit代表符号位，id为正数所以肯定是0

 2.时间戳：占用 41 bit ，精确到毫秒。41位最好可以表示2^41-1毫秒，转化成单位年为 69 年。

3.机器编码：占用10bit，其中高位 5 bit 是数据中心 ID，低位 5 bit 是工作节点 ID，最多可以容纳 1024 个节点。

4.序列号：占用12bit，每个节点每毫秒0开始不断累加，最多可以累加到4095，一共可以产生 4096 个ID。

全局唯一
趋势递增

在Mysql的InnoDB引擎中使用的是聚集索引,由于多数RDBMS使用Btree的数据结构来存储索引数据,在主键的选择上面我们应该尽量使用有序的主键来保证写入性能

单调递增

尽量保证下一个Id一定大于上一个Id,例如事务版本号、IM增量消息、排序等特殊需求

信息安全

如果Id是连续的,恶意用户的扒取工作就非常容易做了,直接按照顺序下载指定URL即可;如果是订单号就更危险了,竞对可以直接知道我们一天的单量.所以在一些应用场景下,需要Id无规则不规则,让竞争对手不好猜

含时间戳

这样就能够在开发中快速了解这个分布式Id的生成时间

上面是生成id的要求，所以雪花算法会有局限性，分布式部署会分配不同的workid，如果workid相同，可能会u导致生成的id重复。雪花算法中有41比特位的dataid，如果生产环境的时间由于某些情况回拨了，可能导致生成的id重复，所以对雪花算法进行改进。

改进：在common包中定义set workid的方法，然后在properties里面配置workerid

* 动态指定sharding jdbc 的雪花算法中的属性work.id属性

* 通过调用System.setProperty()的方式实现,用当前机器ip地址进行hash运算，然后%1024

接下来生成accountid

traffic流量表分表总结：

1.短链的业务场景是用户需要购买不同类型的流量包，这个流量包可以确定用户可以创建多少短链，以及创建短链的有效期是多少。所以这些数据用了traffic表来记录，这个表有一些基本的信息字段，比如创建时间，过期时间，使用次数等等。还有一些比较重要的字段比如商品主键id，account_no（account_no是用户在注册的时候生成唯一的账号标识用自己封装的雪花算法生成）作为一个商品的唯一标识，还有订单号这些都是有唯一索引的。

2.在设计表的时候，粗略估计一下用户数以及数据数量，比如未来三年，有4000w条数据，每个表不能超过1000w的数据，就对traffic表分四张表。之前说过traffic中有accountno这个唯一标识可以作为分表的分片key，可以对account_no对4取模，于是就可以通过sharingjdbc对traffic_0和traffic_1...操作这四张表的数据了。

3.单库下一般使用Mysql自增ID, 但是分库分表后，会出现主键id重复的问题，所以用雪花算法来产生商品的id主键，雪花算法是生成一个64位的数，这个数有一个41位的时间戳，和一个10位的workid（由生产机器决定），在正常情况下能保证单调唯一且递增。

但是小概率下可能还是会有重复，比如分布式部署会分配不同的workid，如果workid相同，可能会导致生成的id重复。雪花算法中有41比特位的dataid就是日期，如果生产环境的时间由于某些情况回拨了，可能导致生成的id重复。所以对雪花算法做了一点改进，根据本机器的ip地址进行hash计算然后取模上1024，就能得到workid的值，这样就更不容易重复了。

4.所以在流量包traffic表中利用唯一标识account_no来进行分表，然后又用改进后的雪花算法生成主键防止分表后的主键id重复的问题。

短链业务和实现

最常见的就是短信发送短链，如果网址太长可能会增加短信的服务费。短链组成就是https协议://短链域名/短链码。还可以对一些商品进行推广以及数据统计。一个长连接可以对应不同的短链接，所以可以用来统计不同渠道推广的短链的点击次数。

重定向

重定向涉及到3xx状态码，访问跳转是301还是302，301和302代表啥意思？

301 是永久重定向
- 会被浏览器硬缓存，第一次会经过短链服务，后续再访问直接从浏览器缓存中获取目标地址
302 是临时重定向
- 不会被浏览器硬缓存，每次都是会访问短链服务
短地址一经生成就不会变化，所以用 301 是同时对服务器压力也会有一定减少
但是如果使用了 301，无法统计到短地址被点击的次数
所以选择302虽然会增加服务器压力，但是有很多数据可以获取进行分析
选择使用302，这个也可以对违规推广的链接进行实时封禁

短链码生成

方案1：数据库自增ID。

利用插入数据库，利用数据库自增id
把自增id转成62进制作为短链码
短链码的长度不固定，随着 id 变大，短链码长度也增长
可以指定从某个长度开始增长，到百亿、千亿数量
是否存在重复: 不重复
但短链码是有序的递增，存在【业务数据安全】问题，比如有一个人今天获得一个短链码，第二天继续获得一个短链码，因为是自增的，他只要计算差值，就能大概知道公司的数据量是多少了

方案2：MD5内容压缩

长链接做md5加密

43E08496,9E5CF455,E6D2D2B3,3407A6D2

加密串查询是否已经生成过短链接
- 如果已经存在，则拼接时间戳再MD5加密，插入数据库
- 如果不存在则把长链接、长链接加密串插入数据库
取MD5后最后1 个 8 位字符串作为短链码
是否存在重复: 存在碰撞（重复）可能
是有损压缩算法，数据量超大情况碰撞概念越大

最终方案

使用MurMurHash（非加密哈希），把长链转化为一个十进制数，它是32bit的能表示最大接近43亿的十进制数，然后把这个十进制数改为62进制。

10进制：1813342104 转62进制：1YIB7i 常规短链码是6~8位数字+大小写字母组合前后是库表位

比如对123abc这个短链，先确定有几个库，比如三个库，0，1，a，四个表0，1，2，3.然后生产这个短链的hash值，对库的数量和表的数量进行取模，来得到相应的库表位。

为了让同一个url生成不同的短链，用作不同平台推广，在原始url前拼接一个时间戳，再去转化为短链，如果这个短链还是重复了，那么就让时间戳加1，然后访问前把这个时间戳去除就可以了。

短链生成方案总结：

1.首先短链的应用场景，最常见的就是短信发送短链，如果网址太长可能会增加短信的服务费。短链组成就是https协议://短链域名/短链码。还可以对一些商品进行推广以及数据统计。一个长连接可以对应不同的短链接，所以可以用来统计不同渠道推广的短链的点击次数，比如在抖音推广使用a短链，在微信推广使用b短链，然后可以统计每个渠道的访问量。

2.短链码生成有很多种方案，

1.比如利用数据库的自增ID来产生，设定一个ID的初始值，比如10w开始自增，然后把这个id值转化为一个62进制的数，他的优点是不会重复，但是缺点也很明显，比如短链的长度不固定，然后存在【业务数据安全】问题，比如有一个人今天获得一个短链码，第二天继续获得一个短链码，因为是自增的，他只要计算差值，就能大概知道公司的数据量是多少了。

2.或者把长连接进行md5加密，但是这样数据一多会有重复的可能。

3.我使用的是MurMurHASH,这是一种非加密的哈希，他把长链转化为一个十进制数，它是32bit的能表示最大接近43亿的十进制数，然后把这个十进制数改为62进制。

比如10进制：1813342104 转62进制：1YIB7i 常规短链码是6~8位数字+大小写字母组合

除了生成的这个短链码，比如123abc，还要对短链码进行处理，用作下一步的分库分表，在短链码的前面加上库位，在最后面加上表位，便于后续对短链的数据进行分库分表。

4.短链生成之后，通过访问短链，先从数据库中找到原始的URL，然后响应http 状态码301或者302重定向到目标地址，然后目标地址对用户响应。

这个过程涉及到对重定向状态码的选择，301 是永久重定向，会被浏览器硬缓存，第一次会经过短链服务，后续再访问直接从浏览器缓存中获取目标地址，302 是临时重定向，不会被浏览器硬缓存，每次都是会访问短链服务。选择301的话会减少服务器的压力，但是无法统计点击短链的次数以及其他数据，所以还是选择302状态码。在代码实现中只要把http响应头中的location属性改成原始的URL，这个属性表示这个短链网址的跳转网址，也就是重定向网址。

5.所以这个就是短链码的应用场景，生成方案的选择，以及短链码跳转的实现。

短链信息表的分库分表

比如 g1.fit/92AEva 的短链码 92AEva，增加库表位

新建三个库，dcloud_0、dcloud_1、dcloud_a里面分别放两个表，short_link_0和short_link_a

然后配置文件

短链码生成有很多种方案，

2.或者把长连接进行md5加密，但是这样数据一多会有重复的可能。

4.短链信息表中主要存储着一些基本的信息，比如id，创建时间，过期时间，标题，短链码和原址URL的值等等。刚才说了通过MurMurHASH根据原始的URL生成一个10进制的数，然后就转为62进制，比如123abc，接下来还要对短链码进行处理，用作下一步的分库分表，在短链码的前面加上库位，在最后面加上表位。

5.比如对123abc这个短链，先确定有几个库，比如三个库，0，1，a，四个表0，1，2，3.然后生产这个短链的hash值，对库的数量和表的数量进行取模，来得到相应的库表位。这样库表为也是为了后续数据能命中相应的库表。然后使用sharingjdbc自定义精准分配策略，把短链码当作分片键，然后通过短链码的首位库位和最后一位表位，获取数据库表。比如0123abc1，这个短链就是进入数据库0，表1中。如果短链码有三个前缀，两个后缀，这样就有六个表。

6.这样的好处就是在数据量增加的时候，扩容避免可以迁移数据，可以直接增加前缀和后缀来增加数据库和表。

#分库分表配置

spring.shardingsphere.datasource.names=ds0,ds1,dsa   #定义数据源

spring.shardingsphere.props.sql.show=true        #显示真实的sql语句

#----------短链，策略：分库+分表--------------

# 先进行水平分库，自定义分片方式

spring.shardingsphere.sharding.tables.short_link.database-strategy.standard.sharding-column=code

spring.shardingsphere.sharding.tables.short_link.database-strategy.standard.precise-algorithm-class-name=net.xdclass.strategy.CustomDBPreciseShardingAlgorithm

# 水平分表策略，自定义策略。   真实库.逻辑表

spring.shardingsphere.sharding.tables.short_link.actual-data-nodes=ds0.short_link,ds1.short_link,dsa.short_link

spring.shardingsphere.sharding.tables.short_link.table-strategy.standard.sharding-column=code

spring.shardingsphere.sharding.tables.short_link.table-strategy.standard.precise-algorithm-class-name=net.xdclass.strategy.CustomTablePreciseShardingAlgorithm

public class CustomDBPreciseShardingAlgorithm implements PreciseShardingAlgorithm<String> {

    /**

     * @param availableTargetNames 数据源集合

     *                             在分库时值为所有分片库的集合 databaseNames

     *                             分表时为对应分片库中所有分片表的集合 tablesNames

     * @param shardingValue        分片属性，包括

     *                             logicTableName 为逻辑表，  short_link

     *                             columnName 分片健（字段），  code

     *                             value 为从 SQL 中解析出的分片健的值 1 186Lgn a

     * @return

     */

    availableTargetNames 数据源集合，这里是ds0，ds1,dsa

    @Override

    public String doSharding(Collection<String> availableTargetNames, PreciseShardingValue<String> shardingValue) {

        //获取短链码第一位，即库位

        String codePrefix = shardingValue.getValue().substring(0, 1);

        for (String targetName : availableTargetNames) {

            //获取库名的最后一位，真实配置的ds

            String targetNameSuffix = targetName.substring(targetName.length() - 1);

            //如果一致则返回

            if (codePrefix.equals(targetNameSuffix)) {

                return targetName;

            }

        }

        //抛异常

        throw new BizException(BizCodeEnum.DB_ROUTE_NOT_FOUND);

    }

}

public class ShardingDBConfig {

    /**

     * 存储数据库位置编号

     */

    private static final List<String> dbPrefixList = new ArrayList<>();

    private static Random random = new Random();

    //配置启用那些库的前缀

    static {

        dbPrefixList.add("0");

        dbPrefixList.add("1");

        dbPrefixList.add("a");

    }

    /**

     * 获取随机的前缀

     * @return

     */

    public static String getRandomDBPrefix(){

        int index = random.nextInt(dbPrefixList.size());

        return dbPrefixList.get(index);

    }

}

短链分组和短链

一个账号有很多分组，每个分组管理多个短链

这里有两张表，短链分组表格很简单，有accout_no唯一标识，分组名字和创建时间

短链表中有基本的信息，名字，过期时间，长链信息，短链状态，是否删除，是否锁定等等。

* 短链-分组

```

CREATE TABLE `link_group` (

  `id` bigint unsigned NOT NULL,

  `title` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '组名',

  `account_no` bigint DEFAULT NULL COMMENT '账号唯一编号',

  `gmt_create` datetime DEFAULT CURRENT_TIMESTAMP,

  `gmt_modified` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

  PRIMARY KEY (`id`)

) ENGINE=InnoDB  DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

```

* 短链

```

CREATE TABLE `short_link` (

  `id` bigint unsigned NOT NULL ,

  `group_id` bigint DEFAULT NULL COMMENT '组',

  `title` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '短链标题',

  `original_url` varchar(1024) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '原始url地址',

  `domain` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '短链域名',

  `code` varchar(16) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '短链压缩码',

  `sign` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '长链的md5码，方便查找',

  `expired` datetime DEFAULT NULL COMMENT '过期时间，长久就是-1',

  `account_no` bigint DEFAULT NULL COMMENT '账号唯一编号',

  `gmt_create` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',

  `gmt_modified` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',

  `del` int unsigned NOT NULL COMMENT '0是默认，1是删除',

  `state` varchar(16) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '状态，lock是锁定不可用，active是可用',

  `link_type` varchar(16) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL COMMENT '链接产品层级：FIRST 免费青铜、SECOND黄金、THIRD钻石',

  PRIMARY KEY (`id`),

  UNIQUE KEY `uk_code` (`code`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

```

短链的CRUD

1.在controller层添加一个比如add(添加分组的方法)，方法中有request对象，前端传入

2.然后controller层添加一个service.add方法，这个方法在service接口中定义

3.完成接口的实现类impl

4.实现类中有一个manager接口层中的add方法，这个add方法直接操作dao层

5.manager实现类中实现crud的操作， 并且注入private LinkGroupMapper linkGroupMapper;

6.LinkGroupMapper extends BaseMapper<LinkGroupDO> 直接用mybatisplus封装的方法

按这个步骤完成所有的crud

分库分表下的多维度查询问题

⼤家在进⾏分库分表的时候应该有碰到⼀个问题，⼀个数据需要根据两种维度进⾏查询，但是我们在进⾏分库分表是只能根据⼀种维度进⾏。⽐如：⽤户购买了商品产⽣了订单，当⽤户⾮常多的时候，我们会选择订单根据下单⽤户的ID进⾏分库分表。但是这⾥⾯存在⼀个问题就是作为卖家要如何查询我卖出的所有订单呢？因为订单是根据⽤户id规则进⾏分库分表的，卖家订单查询接⼝物理上是⽆法⼀次查询出当前的订单的，应该他分别分布在不同的订单库或者订单表中的，有可能我查询⼀个订单列表，需要跨⾮常多的表，这个样性能是⾮常低效的。

冗余双写

https://blog.csdn.net/datuanyuan/article/details/109058337 详细聊聊冗余表数据一致性问题

比如在电商模型中，用户购买了商品产生了订单，订单表根据用户id分表，用户自己查询订单很方便。但是卖家如果要查询所有卖出的订单的话，因为订单是根据用户id分表的，用户非常多，查询一个订单列表，需要跨非常多的表，性能非常低。

同样的，我对短链信息表进行了分库分表，用户可以根据短链信息表进入相应的库表中进行查询，但是如果商家想查询某一分组下的短链，每个短链信息都是不同用户的，也会有跨表操作，所以在创建短链的时候生成两份短链信息。在数据库分别创建用户短链表和商家短链表，商家短链表是给B端用的，访问量少，单表数据可以大一点，用账号的唯一标识account_no分库，用group_id来分表，然后商家查询自己某一组的全部短链数据的时候，就可以通过group_id和account_no作为分片键来查看库表了。

在dcloud_0和1库中分别创建group_code_mapping _0和1的表，这个表和short_link是一样的，只是给b端用的，访问量少，单表数据可以大一点，分库的partitionkey是account_no，分表PartitionKey：group_id.

短链创建：

1.前端传入一个存有短链基本信息的request对象，比如组别group_id，url以及过期时间等等。根据流量包traffic表中查询用户流量包剩余次数，也就是创建短链的剩余次数，如果有足够次数，才能创建短链。

2.这时候并不是直接创建短链，而是用到冗余双写的思想，往商家短链数据库和用户短链数据库写两份数据。这个过程是通过rabbitmq实现的，如果满足创建短链的条件，那么发送一个带有指定路由key的消息，然后消息进入交换机，再进入两个队列，一个是要在用户数据库创建短链信息表的队列，另一个是要在商家短链信息表创建信息的队列。最后在编写两个相应的消费者监听这两个队列的消息，来执行向数据库双写的逻辑。

处理消息失败的问题：在消费者端还设置了重试次数和重试时间，如果有异常消费失败，那么会每隔5秒重试一次，最多能重试四秒。如果重试次数超过了阈值，那么先手工确认一下这个消息，然后把这个消息转发到异常交换机，再到异常队列，然后由异常队列监控服务消费，向相应的研发人员发送短信或者邮件来告警排查问题。

异常交换机底层

其中，异常队列底层是通过RepublishMessageRecoverer这个类实现的，他的初始化可以传入三个参数，rabbittemplate，交换机以及路由的key，当消息重试一定次数后，用特定的routingKey转发到指定的交换机中，

这是由这个类中有一个recover方法实现的，当重试达到阈值时，会告警rupublish失败，然后执行dosend方法，会把这个异常消息转发到异常队列，然后消费服务监听到消息后就能发送短信或者邮件来告警排查问题。

最终一致性

1.这个过程确保了最终一致性，比如一个创建短链的消息，分别到了B端的消费者和C端的消费者，如果C端的消费者消费成功，往用户短链数据库插入了数据，但是C端的消费者消费失败，商家的短链数据库没有数据，那么这次冗余双写只写进去了一个数据库。不过这时候消费失败的那个消息还是在消息队列里面，这个消息重试到一定次数就可以通过异常交换机通过告警服务让开发人员排查原因，解决问题之后确保数据库的一致。这个过程不能实现强一致性，但还是保证了最终的一致。常规情况下，这种情况很少，都会消费成功，但是以防万一也预留了解决方案。

总结：通过发送mq消息队列，来往数据库中写入数据，性能很高，削峰。这是弱一致性，如果需要强一致性的场景，比如往两个数据库插入，要么同时成功，要么同时失败，这样的话就不适用。不过这个业务允许这种场景。

2.或者消费者消费失败，可以写额外的接口来回滚生产者的业务逻辑，就像TCC的cancel。

短链码生成端问题

1.用户提交创建短链的请求，进到controller，然后到service层发送相应的消息。这时候如果在生产者端创建短链，比如AABBCC，这时候要查询短链是否重复创建，比如去数据库中查，这时候查到短链没有被创建，消息也没被消费，数据库中也没插入这个新的短链。这时候如果另一个用户创建短链，由于生成短链冲突，也同样创建了AABBCC的短链，然后查数据库也没重复创建。这时候消息队列中就有两个创建相同短链码的消息了，随后消费肯定会出问题，出现消费失败的情况。

如果在生产者端，出现这种情况也会遇到相同的问题，一个线程往两个数据库写两份数据，两个线程写四份数据，高并发下短链出现重复的情况，最终只会有两份数据写入

加分布式锁问题

方案1：生产者端生成短链码code，加分布式锁 key=短链码code，配置过期时间(加锁失败则重新生成)，这样同一时间段就不会有相同的短链码，然后查询一次数据库或其他存储源，判断是否存在。再发送MQ，C端数据库插入，B端数据库插入，最后解分布式锁（锁过期自动解锁）。

但是这个方案性能很差，用户请求过来，要去加锁查询，然后请求响应给用户，速度会很慢。

方案二：消费者端生成短链码code，生产者发送消息

C端生成，加锁key=code，查询数据库，如果存在，则ver版本递增，重新生成短链码，保存数据库，解分布式锁
B端生成，加锁key=code，查询数据库，如果存在，则ver版本递增，重新生成短链码，保存数据库，解分布式锁
好处，用户请求过来，全部操作比如加锁，查询数据库的操作都是由消费者端完成的

为了让同一个url生成不同的短链，用作不同平台推广，在原始url前拼接一个雪花算法生成的前缀，再去转化为短链，如果这个短链还是重复了，那么就让前缀加1，这个前缀可以看作是版本号，然后访问前把这个前缀去除就可以了。

redis实现分布式锁的问题

比如短链码作为String类型的key，用redis的setnx语句，setnx（key，1），如果这个key不存在，那么设置key，value成功，返回1，如果key已经存在，那么返回0，不成功。比如用if语句判断，if(setnx(key,1)==1) 接下来设置过期时间，在try finnaly中的try语句做相应的业务逻辑，然后业务逻辑执行完后finally删除key。如果if条件不成立就是被阻塞了，自旋使用这个方法。

但是这样存在一个问题，就是多个命令间不是原子性操作，比如第一步setnx设置key，value成功了，第二部操作设置过期时间的时候宕机了，那么这个key一直存在，就造成死锁了。

所以编写一个lua脚本，让redis执行，lua脚本能保证这些命令的原子性，把判断这个key和删除这个key用lua脚本执行，要么全部成功，要么全部失败。整体流程

1.先判断是否有，如没这个key，则设置key-value，配置过期时间，加锁成功

2.如果有这个key,判断value是否是同个账号，是同个账号则返回加锁成功

3.如果不是同个账号则加锁失败

这样就在消费者通过redis+lua脚本实现了分布式锁，保证了操作的原子性。

流量包商品

业务简介：

每个套餐都是一个虚拟商品，没库存限制
免费版是新用户注册即可获得
不同的商品每天限制的创建的条数不一样
用户可以叠加使用多个流量包

流量包商品表有流量包的基本信息，比如详细信息，标题，类型，价格，使用次数，期限等等。

还有流量包订单表

CREATE TABLE `product_order` (

  `id` bigint NOT NULL,

  `product_id` bigint DEFAULT NULL COMMENT '订单类型',

  `product_title` varchar(64) DEFAULT NULL COMMENT '商品标题',

  `product_amount` decimal(16,2) DEFAULT NULL COMMENT '商品单价',

  `product_snapshot` varchar(2048) DEFAULT NULL COMMENT '商品快照',

  `buy_num` int DEFAULT NULL COMMENT '购买数量',

  `out_trade_no` varchar(64) DEFAULT NULL COMMENT '订单唯一标识',

  `state` varchar(11) DEFAULT NULL COMMENT 'NEW 未支付订单,PAY已经支付订单,CANCEL超时取消订单',

  `create_time` datetime DEFAULT NULL COMMENT '订单生成时间',

  `total_amount` decimal(16,2) DEFAULT NULL COMMENT '订单总金额',

  `pay_amount` decimal(16,2) DEFAULT NULL COMMENT '订单实际支付价格',

  `pay_type` varchar(64) DEFAULT NULL COMMENT '支付类型，微信-银行-支付宝',

  `nickname` varchar(64) DEFAULT NULL COMMENT '账号昵称',

  `account_no` bigint DEFAULT NULL COMMENT '用户id',

  `del` int DEFAULT '0' COMMENT '0表示未删除，1表示已经删除',

  `gmt_modified` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',

  `gmt_create` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',

  `bill_type` varchar(32) DEFAULT NULL COMMENT '发票类型：0->不开发票；1->电子发票；2->纸质发票',

  `bill_header` varchar(200) DEFAULT NULL COMMENT '发票抬头',

  `bill_content` varchar(200) DEFAULT NULL COMMENT '发票内容',

  `bill_receiver_phone` varchar(32) DEFAULT NULL COMMENT '发票收票人电话',

  `bill_receiver_email` varchar(200) DEFAULT NULL COMMENT '发票收票人邮箱',

  PRIMARY KEY (`id`),

  UNIQUE KEY `uk_query` (`out_trade_no`,`account_no`) USING BTREE

) ENGINE=InnoDB  DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

防重提交

下单时候，前端按钮重复点击，或者网络原因，用户重复刷新，最后造成订单创建多次

常见方案

1.前端JS控制点击次数，屏蔽点击按钮无法点击，但是前端可以被绕过，前端有限制，后端也需要有限制。

2.数据库或者其他存储增加唯一索引约束，需要想出满足业务需求的唯一索引约束，比如注册的手机号唯一

3.服务端token令牌方式

下单前先获取令牌-存储redis 下单时一并把token提交并检验和删除-lua脚本
分布式情况下，采用Lua脚本进行操作

https://www.cnblogs.com/dw3306/p/9615197.html aop实现自定义注解

自定义注解防重提交

先创建一个自定义注解，自定义注解需要元注解来注解，元注解就是注解注解的注解，最常见的就是，

@Target(ElementType.METHOD)//注解用在方法上

@Retention(RetentionPolicy.RUNTIME)//保留到虚拟机运行时（最多，可通过反射获取）

两种方式

1.令牌形式防重提交

在用户请求的时候可以生成一个token，就用uuid生成一个32位的随机数，提交订单的时候就获取token并且校验一下，然后删除，接下来如果重复请求就找不到token就失败了

2.参数形式防重提交，ip+类+方法方式

用户请求到某个接口，可以拿到类名家方法名加IP，把这些信息当作key存在redis中，并且设置过期时间，比如5秒，如果5秒内有重复请求，那么redis发现有相同的key，就请求失败了。

使用aop+自定义注解可以解耦，比如想在哪个接口那边增加防重提交，只需要在那个方法上增加自己的自定义注解就可以了。把这个注解增加在controller层。

aop就是面向切面编程，开发的步骤就是原始类，额外功能，切入点，组装切面这四个过程。其中后面三个过程可以通过aspect注解定义一个切面类，它的主要作用就是实现额外功能和切入点。实现的额外功能就是刚才说的防重提交的功能，通过@Pointcut和@Around联合注解，pointcut定义切入点就是注解，around注解定义pointcut的方法，然后在这个方法里写入防重的业务逻辑。最后哪个controller接口需要防重提交那就应用这个注解就可以了。

支付中心

支付申请流程：个体工商户，企业，党政机关、事业单位等等可以申请商户号，最后还要和公众号绑定

简单工厂加策略模式

https://blog.csdn.net/yk614294861/article/details/122885700

首先定义了和支付有关方法的strategy抽象策略接口，这个接口有支付、查询订单状态、退款等方法，其中有具体策略类实现，在这里这两个具体的实现类就是支付宝支付方式和微信支付方式的具体实现，里面有对应方式的退款、查询订单状态、以及支付方法。最后还有一个Context上下文类，这个类包含抽象策略类Strategy，负责和具体的策略类交互，他的构造方法的参数就是具体策略的实现类，比如wechatstartegy。

其中简单工厂类的作用是他可以根据参数的不同返回不同类的实例，比如说简单工厂的pay支付方法，传入的参数的支付信息的vo类，这是在创建订单的时候产生的一个类，这个类里面有关于这个订单的基本信息，比如说订单号、订单总金额、支付类型（支付宝还是微信）等等。然后根据支付类型来传入相应的策略类对Context执行初始化，然后执行支付方法。

所以说最后只要根据传入的payinfo信息，使用支付工厂的pay方法，就能创建context对象，并且执行相应方式的支付方法。

而且采用这种设计能减少代码的耦合性，便于以后的修改，比如增加一种支付方式，就是增加一个具体的strategy类。

加密算法我们整体可以分为：可逆加密和不可逆加密，可逆加密又可以分为：对称加密和非对称加密。

不可逆加密：hash算法，MD5，SHA

对称加密：

DES： 全称：Data Encryption Standard，现已被破解

3DES：全称: Triple Data Encryption Algorithm, 暂时未被破解

解释:  3DES 是在 DES 基础算法上的改良，该算法可向下兼容 DES 加密算法，但计算性能不高，暂时还未被破解

AES:  全称：Advanced Encryption Standard，暂未被破解

非对称加密：

流程：

1.接收方生成公私钥对，私钥由接收方保管。

2.接收方将公钥发送给发送方

3.发送方通过公钥对明文加密，得到密文

4.发送方向接收方发送密文

5.接收方通过私钥解密密文，得到明文

对称加密：操作比较简单, 加密速度快, 秘钥简单如果泄露就危险
非对称加密：安全性更高，加解密复杂，但是加解密速度慢

https://blog.csdn.net/qq_45901741/article/details/119223513 https加密

支付宝和微信支付加密方式：采用RSA非对称加密+对称加密混合使用

https传输过程

1.客户端根据https请求到服务端，有公钥和私钥，并且为了防止第三方截取公钥，来冒充代替服务器的公钥，所以先要验证ssl证书（CA机构颁发的数字证书）

2.然后服务器把有资质的crt公钥发给客户端

3.如果客户端验证成功，产生随机的key，通过公钥加密发送给服务端

4.服务端通过自己的私钥解密拿到key

5.然后就用这个key来进行对称加密传输信息

第三方支付平台举例：

提供了两套 RSA 加密

一套是用来保证步骤【统一下单】时的信息安全
另一套是用来保证支付平台回调时的信息安全

具体过程

微信签名和验签的交互图（商户证书（M）和平台证书（W）的使用说明）过程：

1.用户请求到相应的接口，然后用M（私钥）计算签名，微信支付那边用M（公钥)验证签名.。W（私钥）计算签名，然后同步返回给商户，用W（公钥）验签。

2.微信支付成功后要回调通知商户，微信支付用W（私钥）计算签名，商户用W（公钥）验证签名。

其中商户有自己的私钥和公钥，可以使用相应接口下载微信平台的公钥。

微信支付V3开发前工作

1.配置参数

1.申请APPID：

由于微信支付的产品体系全部搭载于微信的社交体系之上，所以直连商户或服务商接入微信支付之前，都需要有一个微信社交载体，该载体对应的ID即为APPID

2.申请mchid：

商户平台有商户号mchid

3.绑定APPID及mchid：

APPID和mchid全部申请完毕后，需要建立两者之间的绑定关系。

直连模式下，APPID与mchid之间的关系为多对多，即一个APPID下可以绑定多个mchid，而一个mchid也可以绑定多个APPID。

2.配置API key

API v3密钥主要用于平台证书解密、回调信息解密，具体使用方式可参见接口规则文档中证书和回调报文解密章节

3.下载并配置商户证书

商户API证书具体使用说明可参见接口规则文档中私钥和证书章节

商户API证书是指由商户申请的，包含商户的商户号、公司名称、公钥信息的证书。

平台证书是指微信支付负责申请的证书，包含微信支付平台标识、公钥信息的证书

商户在调用API时使用自身的私钥签名，微信支付使用商户证书中的公钥来验签。微信支付在相应的报文中使用自身的私钥签名，商户使用平台证书中的公钥来验签名。

apiclient_cert.pem：商户证书，封装了公钥
apiclient_key.pem: 商户证书，封装了私钥

支付策略接口开发

https://pay.weixin.qq.com/wiki/doc/apiv3/open/pay/chapter2_7_2.shtml 开发指引

统一下单接口（unifiedOrder）

natice支付过程：

1.商户的后台系统生成订单，保存到数据库，然后调用微信的统一下单接口，这时候需要传很多参数，比如签名等等。请求过去之后，微信支付系统那边辉生成预支付交易的信息，也会写入相应的数据库，然后返回预支付的code_url，然后后台根据code_url生成相应的二维码连接。

2.接下来这一步是用户和微信支付系统的交互，和我们后台系统没有关系。用户扫二维码，微信支付验证此次交易链接是否有效，以及用户确定支付密码授权，完成支付交易。

3.接下来的过程都是并行处理。用户这边，微信支付完先发送短信和微信消息提醒给用户。商户这边，微信支付会回调通知给商户支付结果，商户返回支付通知接受的情况，这个过程会重试几次。如果没有收到微信支付的回调通知，也可以自己调用查询订单状态的api，来返回支付状态。

4.最后发货

开发具体过程

1.设置参数配置类，WeChatPayConfig

#商户号

pay.wechat.mch-id=1601644442

#公众号id 需要和商户号绑定

pay.wechat.wx-pay-appid=wx5beac15ca207c40c

#商户证书序列号,需要和证书对应

pay.wechat.mch-serial-no=7064ADC5FE84CA2A3DDE71A692E39602DEB96E61

#API V3密钥

pay.wechat.api-v3-key=adP9a0wWjITAbc2oKNP5lfABCxdcl8dy

#商户私钥路径（微信服务端会根据证书序列号，找到证书获取公钥进行解密数据）

pay.wechat.private-key-path=classpath:/cert/apiclient_key.pem

#支付成功页面跳转

pay.wechat.success-return-url=https://xdclass.net

#支付成功，回调通知

pay.wechat.callback-url=http://3xr665.natappfree.cc/api/callback/order/v1/wechat

然后再开发一个PayBeanConfig类加载私钥：apiclient_key.pem

https://github.com/wechatpay-apiv3/wechatpay-apache-httpclient/blob/master/README.md

httpclient依赖，微信在httpclient的基础上封装了签名和验签的过程。可以定时获取微信签名验证器，自动获取微信平台证书（证书里面包括微信平台公钥）, 通过WechatPayHttpClientBuilder构造的HttpClient，会自动的处理签名和验签，并进行证书自动更新。

2.native下单代码编写

https://pay.weixin.qq.com/wiki/doc/apiv3/apis/chapter3_4_1.shtml api文档

    /**

     * 快速验证统一下单接口，验证密钥是否过期

     1.生成商户32位内部订单号

     2.生成post请求，设置传输json数据，json对象中有mchid、out_trade_no、appid、notify_url还有一个amount的json对象

     3.设置请求头和请求体，然后通过封装的httpclient发送请求（其中自带签名，验签过程），返回响应对象，获取响应码和响应体code_url

     * @throws IOException

     */

    @Test

    public void testNativeOrder() throws IOException {

        String outTradeNo = CommonUtil.getStringNumRandom(32);//商户系统内部订单号，只能是数字、大小写字母_-*且在同一个商户号下唯一

        /**

         * {

         * 	"mchid": "1900006XXX",

         * 	"out_trade_no": "native12177525012014070332333",

         * 	"appid": "wxdace645e0bc2cXXX",

         * 	"description": "Image形象店-深圳腾大-QQ公仔",

         * 	"notify_url": "https://weixin.qq.com/",

         * 	"amount": {

         * 		"total": 1,

         * 		"currency": "CNY"

         *        }

         * }

         */

        JSONObject payObj = new JSONObject();

        payObj.put("mchid",payConfig.getMchId());

        payObj.put("out_trade_no",outTradeNo);

        payObj.put("appid",payConfig.getWxPayAppid());

        payObj.put("description","小滴课堂海量数据项目大课");

        payObj.put("notify_url",payConfig.getCallbackUrl());

        //订单总金额，单位为分。

        JSONObject amountObj = new JSONObject();

        amountObj.put("total",100);

        amountObj.put("currency","CNY");

        payObj.put("amount",amountObj);

        //附属参数，可以用在回调

        payObj.put("attach","{\"accountNo\":"+888+"}");

        String body = payObj.toJSONString();

        log.info("请求参数:{}",body);

         //wechatPayClient，之前封装的httpclient，自带签名，验签功能

        StringEntity entity = new StringEntity(body,"utf-8");

        entity.setContentType("application/json");

        HttpPost httpPost = new HttpPost(WechatPayApi.NATIVE_ORDER);

        httpPost.setHeader("Accept","application/json");

        httpPost.setEntity(entity);

        try(CloseableHttpResponse response = wechatPayClient.execute(httpPost)){

            //响应码

            int statusCode = response.getStatusLine().getStatusCode();

            //响应体

            String responseStr = EntityUtils.toString(response.getEntity());

            log.info("下单响应码:{},响应体:{}",statusCode,responseStr);

        }catch (Exception e){

            e.printStackTrace();

        }

    }

2022-05-08 13:47:42.179  INFO 25680 --- [ main] net.xdclass.biz.WechatPayTest: 请求参数:{"amount":{"total":100,"currency":"CNY"},"mchid":"1601644442","out_trade_no":"6BvSHMCXr212tZi6Xsg3mXGCE9NDZ5Bx","appid":"wx5beac15ca207c40c","description":"小滴课堂海量数据项目大课","attach":"{\"accountNo\":888}","notify_url":"http://3xr665.natappfree.cc/api/callback/order/v1/wechat"}

2022-05-08 13:47:42.574  INFO 25680 --- [main] net.xdclass.biz.WechatPayTest            : 下单响应码:200,响应体:{"code_url":"weixin://wxpay/bizpayurl?pr=hqA5QErzz"}

查询订单状态接口（queryPayStatus）

https://pay.weixin.qq.com/wiki/doc/apiv3/apis/chapter3_4_2.shtml 查询订单API

1.获取之前的商户订单号

2.构建URL，使用wechatPayClient执行get方法获取响应json格式的支付状态

    /**

     * 根据商户号订单号查询订单支付状态

     *

     * {"amount":{"payer_currency":"CNY","total":100},"appid":"wx5beac15ca207c40c",

     * "mchid":"1601644442","out_trade_no":"fRAv2Ccpd8GxNEpKAt36X0fdL7WYbn0F",

     * "promotion_detail":[],"scene_info":{"device_id":""},

     * "trade_state":"NOTPAY","trade_state_desc":"订单未支付"}

     *

     * @throws IOException

     */

    @Test

    public void testNativeQuery() throws IOException {

        String outTradeNo = "SDaXxDVeVwmJSxzAkZkP7RYutEGPPkFk";

        String url = String.format(WechatPayApi.NATIVE_QUERY,outTradeNo,payConfig.getMchId());

        HttpGet httpGet = new HttpGet(url);

        httpGet.setHeader("Accept","application/json");

        try(CloseableHttpResponse response = wechatPayClient.execute(httpGet)){

            //响应码

            int statusCode = response.getStatusLine().getStatusCode();

            //响应体

            String responseStr = EntityUtils.toString(response.getEntity());

            log.info("查询响应码:{},响应体:{}",statusCode,responseStr);

        }catch (Exception e){

            e.printStackTrace();

        }

    }

关闭订单接口（closeOrder）

https://pay.weixin.qq.com/wiki/doc/apiv3/apis/chapter3_4_3.shtml api文档

以下情况需要调用关单接口：

1、商户订单支付失败需要生成新单号重新发起支付，要对原订单号调用关单，避免重复支付；

2、系统下单后，用户支付超时，系统退出不再受理，避免用户继续，请调用关单接口。

请求URL： https://api.mch.weixin.qq.com/v3/pay/transactions/out-trade-no/{out_trade_no}/close

请求方式： POST

1.获取订单号

2.将商户号machid设置到body里，设置url，然后用wechatPayClient发送post请求

3.无返回值，只许看状态码204就是成功了。

    @Test

    public void testNativeCloseOrder() throws IOException {

        String outTradeNo = "SDaXxDVeVwmJSxzAkZkP7RYutEGPPkFk";

        JSONObject payObj = new JSONObject();

        payObj.put("mchid",payConfig.getMchId());

        String body = payObj.toJSONString();

        log.info("请求参数:{}",body);

        //将请求参数设置到请求对象中

        StringEntity entity = new StringEntity(body,"utf-8");

        entity.setContentType("application/json");

        String url = String.format(WechatPayApi.NATIVE_CLOSE_ORDER,outTradeNo);

        HttpPost httpPost = new HttpPost(url);

        httpPost.setHeader("Accept","application/json");

        httpPost.setEntity(entity);

        try(CloseableHttpResponse response = wechatPayClient.execute(httpPost)){

            //响应码

            int statusCode = response.getStatusLine().getStatusCode();

            log.info("关闭订单响应码:{},无响应体",statusCode);

        }catch (Exception e){

            e.printStackTrace();

        }

    }

退款接口（refund）

当交易发生之后一年内，由于买家或者卖家的原因需要退款时，卖家可以通过退款接口将支付金额退还给买家，微信支付将在收到退款请求并且验证成功之后，将支付款按原路退还至买家账号上。

注意：

1、交易时间超过一年的订单无法提交退款

2、微信支付退款支持单笔交易分多次退款（不超50次），多次退款需要提交原支付订单的商户订单号和设置不同的退款单号。申请退款总金额不能超过订单金额。一笔退款失败后重新提交，请不要更换退款单号，请使用原商户退款单号

3、错误或无效请求频率限制：6qps，即每秒钟异常或错误的退款申请请求不超过6次

4、每个支付订单的部分退款次数不能超过50次

5、如果同一个用户有多笔退款，建议分不同批次进行退款，避免并发退款导致退款失败

6、申请退款接口的返回仅代表业务的受理情况，具体退款是否成功，需要通过退款查询接口获取结果

7、一个月之前的订单申请退款频率限制为：5000/min

8、同一笔订单多次退款的请求需相隔1分钟

@Test

/*

     1.获取商户32位内部订单号，生成退款订单号

     2.生成post请求，设置传输json数据，json对象中有mchid、out_trade_no、appid、notify_url还有一个amount的json对象

     3.设置请求头和请求体，然后通过封装的httpclient发送请求（其中自带签名，验签过程），返回响应对象，获取响应码和响应体code_url

     */

public void testNativeRefundOrder() throws IOException {

    String outTradeNo = "uP3FnON8ts7aab7E25jO9flM1QFf37Hb";

    String refundNo = CommonUtil.getStringNumRandom(32);

    // 请求body参数

    JSONObject refundObj = new JSONObject();

    //订单号

    refundObj.put("out_trade_no", outTradeNo);

    //退款单编号，商户系统内部的退款单号，商户系统内部唯一，

    // 只能是数字、大小写字母_-|*@ ，同一退款单号多次请求只退一笔

    refundObj.put("out_refund_no", refundNo);

    refundObj.put("reason","商品已售完");

    refundObj.put("notify_url", payConfig.getCallbackUrl());

    JSONObject amountObj = new JSONObject();

    //退款金额

    amountObj.put("refund", 10);

    //实际支付的总金额

    amountObj.put("total", 100);

    amountObj.put("currency", "CNY");

    refundObj.put("amount", amountObj);

    String body = refundObj.toJSONString();

    log.info("请求参数:{}",body);

    StringEntity entity = new StringEntity(body,"utf-8");

    entity.setContentType("application/json");

    HttpPost httpPost = new HttpPost(WechatPayApi.NATIVE_REFUND_ORDER);

    httpPost.setHeader("Accept","application/json");

    httpPost.setEntity(entity);

    try(CloseableHttpResponse response = wechatPayClient.execute(httpPost)){

        //响应码

        int statusCode = response.getStatusLine().getStatusCode();

        //响应体

        String responseStr = EntityUtils.toString(response.getEntity());

        log.info("申请订单退款响应码:{},响应体:{}",statusCode,responseStr);

    }catch (Exception e){

        e.printStackTrace();

    }

}

回调验签

适用对象： 直连商户

请求方式：POST

回调URL：该链接是通过基础下单接口中的请求参数“notify_url”来设置的，要求必须为https地址。请确保回调URL是外部可正常访问的，且不能携带后缀参数，否则可能导致商户无法接收到微信的回调通知信息。回调URL示例：“https://pay.weixin.qq.com/wxpay/pay.action”

https://pay.weixin.qq.com/wiki/doc/apiv3/apis/chapter3_4_5.shtml
https://pay.weixin.qq.com/wiki/doc/apiv3/wechatpay/wechatpay4_1.shtml
同样的通知可能会多次发送给商户系统，商户系统必须能够正确处理重复的通知
推荐的做法是，当商户系统收到通知进行处理时，先检查对应业务数据的状态，并判断该通知是否已经处理，如果未处理，则再进行处理；如果已处理，则直接返回结果成功。
在对业务数据进行状态检查和处理之前，要采用数据锁进行并发控制，以避免函数重入造成的数据混乱。
如果在所有通知频率后没有收到微信侧回调，商户应调用查询订单接口确认订单状态
确保回调URL是外部可正常访问的，且不能携带后缀参数
编码实战-核心流程操作
用户支付完成后，微信会把相关支付结果和用户信息发送给商户，商户需要接收处理该消息，并返回应答。
- 首先要从微信回调的请求中获取响应的报文，比如HTTP头应答时间戳、应答随机串、平台证书序列号，其中要先检查平台证书序列号和商户所持有的是否一致，然后再然按一定规则构造验签名串，就是把前面获取的信息按照顺序组合。
- (HTTP头Wechatpay-Timestamp 中的应答时间戳、HTTP头Wechatpay-Nonce 中的应答随机串，微信支付的平台证书序列号位于HTTP头Wechatpay-Serial。验证签名前，请商户先检查序列号是否跟商户当前所持有的微信支付平台证书的序列号一致。如果不一致，请重新获取证书。否则，签名的私钥和证书不匹配，将无法成功验证签名。)
- 验证签名（确保是微信传输过来的），通过微信封装的方法验证。
- 因为支付结果通知是以POST 方法访问商户设置的通知url，通知的数据以JSON 格式通过请求主体（BODY）传输。通知的数据包括了加密的支付结果详情，所以需要解密（AES对称解密出原始数据）主体(body)信息。
- 处理业务逻辑（更新订单状态，发放相应商品（流量包））
- 响应请求，返回一个map格式的数据

@Controller

@RequestMapping("/api/callback/order/v1/")

@Slf4j

public class PayCallbackController {

    @Autowired

    private WechatPayConfig wechatPayConfig;

    @Autowired

    private ProductOrderService productOrderService;

    @Autowired

    private ScheduledUpdateCertificatesVerifier verifier;

    /**

     * * 获取报文()

     * <p>

     * * 验证签名（确保是微信传输过来的）

     * <p>

     * * 解密（AES对称解密出原始数据）

     * <p>

     * * 处理业务逻辑

     * <p>

     * * 响应请求

{

    "code": "FAIL",

    "message": "失败"

}   返回map类型

     * @param request

     * @param response

     * @return

     */

    @RequestMapping("wechat")

    @ResponseBody

    public Map<String, String> wehcatPayCallback(HttpServletRequest request, HttpServletResponse response) {

        //获取报文

        String body = getRequestBody(request);

        //随机串

        String nonceStr = request.getHeader("Wechatpay-Nonce");

        //微信传递过来的签名

        String signature = request.getHeader("Wechatpay-Signature");

        //证书序列号（微信平台）

        String serialNo = request.getHeader("Wechatpay-Serial");

        //时间戳

        String timestamp = request.getHeader("Wechatpay-Timestamp");

        //构造签名串

        //应答时间戳\n

        //应答随机串\n

        //应答报文主体\n

        String signStr = Stream.of(timestamp, nonceStr, body).collect(Collectors.joining("\n", "", "\n"));

        Map<String, String> map = new HashMap<>(2);

        try {

            //验证签名是否通过

            boolean result = verifiedSign(serialNo, signStr, signature);

            if(result){

                //解密数据

                String plainBody = decryptBody(body);

                log.info("解密后的明文:{}",plainBody);

                Map<String, String> paramsMap = convertWechatPayMsgToMap(plainBody);

                //处理业务逻辑

                productOrderService.processOrderCallbackMsg(ProductOrderPayTypeEnum.WECHAT_PAY,paramsMap);

                //响应微信

                map.put("code", "SUCCESS");

                map.put("message", "成功");

            }

        } catch (Exception e) {

            log.error("微信支付回调异常:{}", e);

        }

        return map;

    }

    /**

     * 转换body为map

     * @param plainBody

     * @return

     */

    private Map<String,String> convertWechatPayMsgToMap(String plainBody){

        Map<String,String> paramsMap = new HashMap<>(2);

        JSONObject jsonObject = JSONObject.parseObject(plainBody);

        //商户订单号

        paramsMap.put("out_trade_no",jsonObject.getString("out_trade_no"));

        //交易状态

        paramsMap.put("trade_state",jsonObject.getString("trade_state"));

        //附加数据

        paramsMap.put("account_no",jsonObject.getJSONObject("attach").getString("accountNo"));

        return paramsMap;

    }

    /**

     * 解密body的密文

     *

     * "resource": {

     *         "original_type": "transaction",

     *         "algorithm": "AEAD_AES_256_GCM",

     *         "ciphertext": "",

     *         "associated_data": "",

     *         "nonce": ""

     *     }

     *

     * @param body

     * @return

     */

    private String decryptBody(String body) throws UnsupportedEncodingException, GeneralSecurityException {

        AesUtil aesUtil = new AesUtil(wechatPayConfig.getApiV3Key().getBytes("utf-8"));

        JSONObject object = JSONObject.parseObject(body);

        JSONObject resource = object.getJSONObject("resource");

        String ciphertext = resource.getString("ciphertext");

        String associatedData = resource.getString("associated_data");

        String nonce = resource.getString("nonce");

        return aesUtil.decryptToString(associatedData.getBytes("utf-8"),nonce.getBytes("utf-8"),ciphertext);

    }

    /**

     * 验证签名

     *

     * @param serialNo  微信平台-证书序列号

     * @param signStr   自己组装的签名串

     * @param signature 微信返回的签名

     * @return

     * @throws UnsupportedEncodingException

     */

    private boolean verifiedSign(String serialNo, String signStr, String signature) throws UnsupportedEncodingException {

        return verifier.verify(serialNo, signStr.getBytes("utf-8"), signature);

    }

    /**

     * 读取请求数据流

     *

     * @param request

     * @return

     */

    private String getRequestBody(HttpServletRequest request) {

        StringBuffer sb = new StringBuffer();

        try (ServletInputStream inputStream = request.getInputStream();

             BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));

        ) {

            String line;

            while ((line = reader.readLine()) != null) {

                sb.append(line);

            }

        } catch (IOException e) {

            log.error("读取数据流异常:{}", e);

        }

        return sb.toString();

    }

}

验签完的订单状态更新和流量包发放

首先要从微信回调的请求中获取响应的报文，比如HTTP头应答时间戳、应答随机串、平台证书序列号，其中要先检查平台证书序列号和商户所持有的是否一致，然后再然按一定规则构造验签名串，就是把前面获取的信息按照顺序组合。
(HTTP头Wechatpay-Timestamp 中的应答时间戳、HTTP头Wechatpay-Nonce 中的应答随机串，微信支付的平台证书序列号位于HTTP头Wechatpay-Serial。验证签名前，请商户先检查序列号是否跟商户当前所持有的微信支付平台证书的序列号一致。如果不一致，请重新获取证书。否则，签名的私钥和证书不匹配，将无法成功验证签名。)
验证签名（确保是微信传输过来的），通过微信封装的方法验证。
因为支付结果通知是以POST 方法访问商户设置的通知url，通知的数据以JSON 格式通过请求主体（BODY）传输。通知的数据包括了加密的支付结果详情，所以需要解密（AES对称解密出原始数据）主体(body)信息。
处理业务逻辑（更新订单状态，发放相应商品（流量包））
响应请求，返回一个map格式的数据

其中处理业务逻辑就是订单状态更新和流量包发放

方案1:

1-更新订单：数据库IO
2-RPC发放流量包：网络IO+数据库IO
3-最终才响应给微信

问题：更新订单的时候和rpc发放流量包的时候都要用数据库io，因为微信是定时通知的，如果处理的慢，微信会发送很多通知。

还要分布式事务问题，1，2两步其中有一个失败就会造成数据不一致，比如发了流量包，订单还是未支付。

总结1.性能问题2.分布式事务问题

方案2：

订单服务经过交换机直接进入队列冗余双写，然后响应微信支付接口，再消费消息-更新数据库订单状态和发放流量包

性能：微信给订单服务请求，订单服务发送消息，然后响应微信，比之前先用数据库更新订单和发放流量包接口性能提高了很多

分布式：消费的时候做好异常处理，确保两个消息都被消费成功。

方案3：

折中方案，接收微信回调通知更新数据库，发送新增流量包MQ消息，响应微信，再消费流量包消息

微信推一个通知过来，商品订单服务先直接更新订单状态，因为调用同一个微服务中的方法，不用rpc调用。但是发放流量包要用到账号微服务，这时候商品订单服务先直接更新订单状态后，再发送一个消息给账号微服务消费，用于发放流量包。因为第一步更新订单状态和给发放流量包的账号微服务发送消息这两个方法是在同一个service，只要开启一个事务就可以了。

方案二有两个队列，性能最好，也能解决分布式事务问题，所以采用方案二。

商品微服务用mq：

给用户新增发送流量包

* 由业务性能决定哪种方式,空间换时间，时间换空间

* 方式一：商品信息/订单信息 可以由消费者那边 远程调用feign进行获取，多了一次开销

* 方式二：商品信息进行快照存储到订单，支付通知回调，组装消息体进行发送（空间换时间，推荐）

    /**

     * 处理微信回调通知

     注意点1：；需要开启事务，transactional

     *2.content存储订单的基本必备内容

     3.消息用订单号做唯一标识存储在redis防止重复投递

     * @param wechatPay

     * @param paramsMap

     */

    @Override

    @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)

    public JsonData processOrderCallbackMsg(ProductOrderPayTypeEnum payType, Map<String, String> paramsMap) {

        //获取商户订单号

        String outTradeNo = paramsMap.get("out_trade_no");

        //交易状态

        String tradeState = paramsMap.get("trade_state");

        Long accountNo = Long.valueOf(paramsMap.get("account_no"));

        ProductOrderDO productOrderDO = productOrderManager.findByOutTradeNoAndAccountNo(outTradeNo, accountNo);

        Map<String, Object> content = new HashMap<>(4);

        content.put("outTradeNo", outTradeNo);

        content.put("buyNum", productOrderDO.getBuyNum());

        content.put("accountNo", accountNo);

        content.put("product", productOrderDO.getProductSnapshot());

        //构建消息

        EventMessage eventMessage = EventMessage.builder()

                .bizId(outTradeNo)

                .accountNo(accountNo)

                .messageId(outTradeNo)

                .content(JsonUtil.obj2Json(content))

                .eventMessageType(EventMessageType.PRODUCT_ORDER_PAY.name())

                .build();

        if (payType.name().equalsIgnoreCase(ProductOrderPayTypeEnum.ALI_PAY.name())) {

            //支付宝支付 TODO

        } else if (payType.name().equalsIgnoreCase(ProductOrderPayTypeEnum.WECHAT_PAY.name())) {

           if ("SUCCESS".equalsIgnoreCase(tradeState)) {

                //如果key不存在，则设置成功，返回true

                //如果为空就set值，并返回1

                //如果存在(不为空)不进行操作，并返回0

                Boolean flag = redisTemplate.opsForValue().setIfAbsent(outTradeNo, "OK", 3, TimeUnit.DAYS);

                if (flag) {

                    rabbitTemplate.convertAndSend(rabbitMQConfig.getOrderEventExchange(),

                            rabbitMQConfig.getOrderUpdateTrafficRoutingKey(), eventMessage);

                    return JsonData.buildSuccess();

                }

            }

        }

        return JsonData.buildResult(BizCodeEnum.PAY_ORDER_CALLBACK_NOT_SUCCESS);

    }

消费者：监听到消息执行下面逻辑

    /**

     * 处理订单相关消息

     * @param message

     */

    @Override

    public void handleProductOrderMessage(EventMessage eventMessage) {

        String messageType = eventMessage.getEventMessageType();

        try{

            if(EventMessageType.PRODUCT_ORDER_NEW.name().equalsIgnoreCase(messageType)){

                //关闭订单

                this.closeProductOrder(eventMessage);

            } else if(EventMessageType.PRODUCT_ORDER_PAY.name().equalsIgnoreCase(messageType)){

                //订单已经支付，更新订单状态

                String outTradeNo = eventMessage.getBizId();

                Long accountNo = eventMessage.getAccountNo();

                int rows = productOrderManager.updateOrderPayState(outTradeNo,accountNo,

                       ProductOrderStateEnum.PAY.name(),ProductOrderStateEnum.NEW.name());

                log.info("订单更新成功:rows={},eventMessage={}",rows,eventMessage);

            }

        }catch (Exception e){

            log.error("订单消费者消费失败:{}",eventMessage);

            throw new BizException(BizCodeEnum.MQ_CONSUME_EXCEPTION);

        }

    }

账号微服务来做流量包权益更新

1.先完成流量包模块的crud

流量包分页查询和查询具体流量包的信息（略）

流量包减少（reduce）

    /**

     * * 查询用户全部可用流量包

     * * 遍历用户可用流量包

     *   * 判断是否更新-用日期判断

     *     * 没更新的流量包后加入【待更新集合】中

     *       * 增加【今天剩余可用总次数】

     *     * 已经更新的判断是否超过当天使用次数

     *       * 如果没超过则增加【今天剩余可用总次数】

     *       * 超过则忽略

     *

     * * 更新用户今日流量包相关数据

     * * 扣减使用的某个流量包使用次数

     * @param useTrafficRequest

     * @return

     */

    @Override

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRED)

    public JsonData reduce(UseTrafficRequest trafficRequest) {

        Long accountNo = trafficRequest.getAccountNo();

        //处理流量包，筛选出未更新流量包，当前使用的流量包

        UseTrafficVO useTrafficVO = processTrafficList(accountNo);

        log.info("今天可用总次数:{},当前使用流量包:{}",useTrafficVO.getDayTotalLeftTimes(),useTrafficVO.getCurrentTrafficDO());

        if(useTrafficVO.getCurrentTrafficDO() == null){

            return JsonData.buildResult(BizCodeEnum.TRAFFIC_REDUCE_FAIL);

        }

        log.info("待更新流量包列表:{}",useTrafficVO.getUnUpdatedTrafficIds());

        if(useTrafficVO.getUnUpdatedTrafficIds().size()>0){

            //更新今日流量包

            trafficManager.batchUpdateUsedTimes(accountNo,useTrafficVO.getUnUpdatedTrafficIds());

        }

        //先更新，再扣减当前使用的流量包

        int rows = trafficManager.addDayUsedTimes(accountNo,useTrafficVO.getCurrentTrafficDO().getId(),1);

        TrafficTaskDO trafficTaskDO = TrafficTaskDO.builder().accountNo(accountNo).bizId(trafficRequest.getBizId())

                .useTimes(1).trafficId(useTrafficVO.getCurrentTrafficDO().getId())

                .lockState(TaskStateEnum.LOCK.name()).build();

        trafficTaskManager.add(trafficTaskDO);

        if(rows != 1){

            throw new BizException(BizCodeEnum.TRAFFIC_REDUCE_FAIL);

        }

        //往redis设置下总流量包次数，短链服务那边递减即可； 如果有新增流量包，则删除这个key

        long leftSeconds = TimeUtil.getRemainSecondsOneDay(new Date());

        String totalTrafficTimesKey = String.format(RedisKey.DAY_TOTAL_TRAFFIC,accountNo);

        redisTemplate.opsForValue().set(totalTrafficTimesKey,

                useTrafficVO.getDayTotalLeftTimes()-1,leftSeconds, TimeUnit.SECONDS);

        EventMessage trafficUseEventMessage = EventMessage.builder().accountNo(accountNo).bizId(trafficTaskDO.getId() + "")

                .eventMessageType(EventMessageType.TRAFFIC_USED.name()).build();

        //发送延迟消息，用于异常回滚

        rabbitTemplate.convertAndSend(rabbitMQConfig.getTrafficEventExchange(),

                rabbitMQConfig.getTrafficReleaseDelayRoutingKey(),trafficUseEventMessage);

        return JsonData.buildSuccess();

    }

2.配置mq，用来监听商品微服务发送过来的消息，消费者执行业务逻辑

监听三个队列

@RabbitListener(queuesToDeclare = {

        @Queue("order.traffic.queue"),

        @Queue("traffic.free_init.queue"),

        @Queue("traffic.release.queue")

})

//1.订单已经支付的情况：因为在商品订单服务发送消息的时候，把订单的一些基本信息放在了消息中，所以先获取消息中订单的基本信息，构建流量包对象，插入数据库

//2.traffic.free_init.queue ,这是在用户注册成功的时候发放免费流量包的逻辑* 注册成功发送MQ消息，响应用户，监听到traffic.free_init.queue中的消息后，然后同样在handleTrafficMessage中处理，构建免费流量包对象，然后插入数据库

//3.

* 消费端处理发放免费流量包

    @Override

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRED)

    public void handleTrafficMessage(EventMessage eventMessage) {

        Long accountNo = eventMessage.getAccountNo();

        String messageType = eventMessage.getEventMessageType();

        if(EventMessageType.PRODUCT_ORDER_PAY.name().equalsIgnoreCase(messageType)){

            //订单已经支付，新增流量

            String content = eventMessage.getContent();

            Map<String, Object> orderInfoMap = JsonUtil.json2Obj(content,Map.class);

            //还原订单商品信息

            String outTradeNo = (String)orderInfoMap.get("outTradeNo");

            Integer buyNum = (Integer)orderInfoMap.get("buyNum");

            String productStr = (String) orderInfoMap.get("product");

            ProductVO productVO = JsonUtil.json2Obj(productStr, ProductVO.class);

            log.info("商品信息:{}",productVO);

            //流量包有效期

            LocalDateTime expiredDateTime = LocalDateTime.now().plusDays(productVO.getValidDay());//过期时间

            Date date = Date.from(expiredDateTime.atZone(ZoneId.systemDefault()).toInstant());//转化成data数据类型

            //构建流量包对象

            TrafficDO trafficDO = TrafficDO.builder()

                    .accountNo(accountNo)

                    .dayLimit(productVO.getDayTimes() * buyNum)

                    .dayUsed(0)

                    .totalLimit(productVO.getTotalTimes())

                    .pluginType(productVO.getPluginType())

                    .level(productVO.getLevel())

                    .productId(productVO.getId())

                    .outTradeNo(outTradeNo)

                    .expiredDate(date).build();

            int rows = trafficManager.add(trafficDO);

            log.info("消费消息新增流量包:rows={},trafficDO={}",rows,trafficDO);

            //新增流量包，应该删除这个key

            String totalTrafficTimesKey = String.format(RedisKey.DAY_TOTAL_TRAFFIC,accountNo);

            redisTemplate.delete(totalTrafficTimesKey);

        }else if(EventMessageType.TRAFFIC_FREE_INIT.name().equalsIgnoreCase(messageType)){

            //发放免费流量包

            Long productId = Long.valueOf(eventMessage.getBizId());

            JsonData jsonData = productFeignService.detail(productId);

            ProductVO productVO = jsonData.getData(new TypeReference<ProductVO>(){});

            //构建流量包对象

            TrafficDO trafficDO = TrafficDO.builder()

                    .accountNo(accountNo)

                    .dayLimit(productVO.getDayTimes())

                    .dayUsed(0)

                    .totalLimit(productVO.getTotalTimes())

                    .pluginType(productVO.getPluginType())

                    .level(productVO.getLevel())

                    .productId(productVO.getId())

                    .outTradeNo("free_init")

                    .expiredDate(new Date())

                    .build();

            trafficManager.add(trafficDO);

        }

        else if(EventMessageType.TRAFFIC_USED.name().equalsIgnoreCase(messageType)){

            //流量包使用，检查是否成功使用

            //检查task是否存在

            //检查短链是否成功

            //如果不成功，则恢复流量包

            //删除task (也可以更新task状态，定时删除就行)

            Long trafficTaskId = Long.valueOf(eventMessage.getBizId());

            TrafficTaskDO trafficTaskDO = trafficTaskManager.findByIdAndAccountNo(trafficTaskId, accountNo);

            //非空且锁定

            if(trafficTaskDO !=null && trafficTaskDO.getLockState().equalsIgnoreCase(TaskStateEnum.LOCK.name())){

                JsonData jsonData = shortLinkFeignService.check(trafficTaskDO.getBizId());

                if(jsonData.getCode()!= 0 ){

                    log.error("创建短链失败，流量包回滚");

                    String useDateStr =  TimeUtil.format(trafficTaskDO.getGmtCreate(),"yyyy-MM-dd");

                  trafficManager.releaseUsedTimes(accountNo,trafficTaskDO.getTrafficId(),1,useDateStr);

                    //恢复流量包，应该删除这个key（也可以让这个key递增）

                    String totalTrafficTimesKey = String.format(RedisKey.DAY_TOTAL_TRAFFIC,accountNo);

                    redisTemplate.delete(totalTrafficTimesKey);

                }

                //多种方式处理task，不立刻删除，可以更新状态，然后定时删除也行

                trafficTaskManager.deleteByIdAndAccountNo(trafficTaskId,accountNo);

            }

        }

    }

消息队列重复投递

比如说生产者会发送多个消息，这时候消费端处理业务消息要保证幂等性。以微信支付为例，在微信支付成功后，微信支付会多次回调商家设置的回调地址，然后比如商家收到这个请求后使用消息队列来处理相应的业务逻辑，这时候微信多次请求，生产者就会发送多次消息，这时候可以将生产者设置唯一标识的key值，比如在这个场景里就可以用业务唯一标识订单号来做幂等处理。这里可以用数据库或者redis来记录唯一标识。如果用数据库来记录，那么这样会增加表记录，还需要用定时任务来清除历史的表记录，增加了数据库的io操作。另一种方法是用redis设置key，value来配置过期时间。key是订单号，value可以随便设置。如果在过期时间内，能查到相应的key，value，那么就是重复投递，不处理。

消息队列重复消费

在生产者端防止消息重复投递，在业务端也需要去重，举个例子，就是购买虚拟物品比如会员，重复消费就会增加会员的天数。这时候可以用订单号在数据库中作唯一索引或者其他的特定标识在数据库中作唯一索引，这样就不会被重复消费。

例如：之前在生产者端做了防止重复投递的处理，但是还是不能保证100%防止，所以在消费者端也要进行处理。发放流量包业务中，因为traffic表用的是accountno作为分片键来分表的，能进入同个库表所以就用唯一索引。微信支付发送多次消息，outTradeNo肯定是唯一的，所以就用这个作为唯一索引，所以只能插入一次。免费流量包也有唯一标识free_init ，用来做唯一索引。

MQ消费端幂等性保障

分库分表情况下-同个accountNo路由到同个库表
付费流量包
- 采用订单号做唯一索引（accountNo + outTradeNo）
免费流量包
- 采用特定标识做唯一索引（ accountNo + free_init ）

rabbitmq的可靠性投递

⽣产者-->交换机->队列->消费者

通过两个的点控制消息的可靠性投递

⽣产者到交换机

通过confirmCallback确认模式，在创建connect工厂的时候可以设置这个confirmCallback确定模式，其中confirmCallback接口中有一个confirm方法，里面有一个CorrelationData对象，CorrelationData 对象内部只有一个 id 属性，用来表示当前消息唯一性。消息只要被 rabbitmq 交换机接收到就会执行 confirmCallback。但是这时候并不能确定投递到队列。

交换机到队列

通过returnCallback，同样创建 ConnectionFactory 到时候需要设置 PublisherReturns(true) 模式，这样如果未能投递到目标 queue 里将调用 returnCallback ，可以记录下详细到投递数据，定期的巡检或者自动纠错都需要这些数据。

建议

开启消息确认机制以后，保证了消息的准确送达，但由于频繁的确认交互， rabbitmq 整体效率变低，吞吐量

下降严重，不是⾮常重要的消息真⼼不建议⽤消息确认机制。

rabbit持久化

https://blog.csdn.net/java123456111/article/details/124023038

流量包的过期删除问题和方案

需求：

流量包更新维护需求

付费流量包：通过购买，然后每天都是有一定的使用次数
免费流量包：业务为了拉新，鼓励新用户注册，赠送一个免费流量包，每天允许有一定次免费创建短链的次数

无论是免费还是付费流量包，每天都有使用次数，并且需要更新使用次数

1.定时任务-xxljob

每天用xxl-job调度中心定时执行账号微服务中的流量包更新服务

问题：用户量大，会有更新延迟。

需求
- 用户购买的流量包都是有时间限制，过期的流量包需要删除
- 逻辑删除、物理删除、或者转移到日志文件归档都行
- 我们这边直接使用物理删除，比数据过多
- 每天定时执行delete FROM traffic where expired_date <= now()
解决方式
- 使用定时任务删除
- 使用场景
  - 某个时间定时处理某个任务、发邮件、短信等
- 消息提醒、订单通知、统计报表（最多）系

XXL-JOB

大众点评的员工徐雪里在15年发布的分布式任务调度平台，是轻量级的分布式任务调度框架，目标是开发迅速、简单、清理、易扩展; 老版本是依赖quartz的定时任务触发，在v2.1.0版本开始移除quartz依赖
地址：https://www.xuxueli.com/xxl-job/
常规对比图

对比项	XXL-JOB	elastic-job
并行调度	调度系统多线程并行	任务分片的方式并行
弹性扩容	使用Quartz基于数据库分布式功能	通过zookeeper保证
高可用	通过DB锁保证	通过zookeeper保证
阻塞策略	单机串行/丢弃后续的调度/覆盖之前的调度	执行超过zookeeper的session timeout时间的话，会被清除，重新进行分片
动态分片策略	以执行器为维度进行分片、支持动态的扩容	平均分配/作业名hash分配/自定义策略
失败处理策略	失败告警/失败重试	执行完毕后主动获取未分配分片任务服务器下线后主动寻找可以用的服务器执行任务
监控	支持	支持
日志	支持	支持

如何选择哪一个分布式任务调度平台
- XXL-Job和Elastic-Job都具有广泛的用户基础和完善的技术文档，都可以满足定时任务的基本功能需求
- xxl-job侧重在业务实现简单和管理方便，容易学习，失败与路由策略丰富, 推荐使用在用户基数相对较少，服务器的数量在一定的范围内的场景下使用
- elastic-job关注的点在数据，添加了弹性扩容和数据分片的思路，更方便利用分布式服务器的资源, 但是学习难度较大，推荐在数据量庞大，服务器数量多的时候使用

搭建XXL-Job相关环境

创建数据库脚本

* xxl_job_group：执行器信息表，用于维护任务执行器的信息

* xxl_job_info：调度扩展信息表，主要是用于保存xxl-job的调度任务的扩展信息，比如说像任务分组、任务名、机器的地址等等

* xxl_job_lock：任务调度锁表

* xxl_job_log：日志表，主要是用在保存xxl-job任务调度历史信息，像调度结果、执行结果、调度入参等等

* xxl_job_log_report：日志报表，会存储xxl-job任务调度的日志报表，会在调度中心里的报表功能里使用到

* xxl_job_logglue：任务的GLUE日志，用于保存GLUE日志的更新历史变化，支持GLUE版本的回溯功能

* xxl_job_registry：执行器的注册表，用在维护在线的执行器与调度中心的地址信息

* xxl_job_user：系统的用户表

部署XXL-Job服务端
客户端项目添加依赖

界面：

运行报表

以图形化来展示了整体的任务执行情况
- 任务数量：能够看到调度中心运行的任务数量
- 调度次数：调度中心所触发的调度次数
- 执行器数量：在整个调度中心中，在线的执行器数量有多少

任务管理

这里是配置执行任务/路由策略/Cron/JobHandler等

调度日志

这里是查看调度的日志，根据日志来查看任务具体的执行情况是怎样的

执行器管理

这里是配置执行器，等待执行器启动的时候都会被调度中心监听加入到地址列表

Redis过期key淘汰策略

主动策略：给过期的key加定时器，当时间到达过期时间的时候会自动删除key，等于cpu不友好，会占用很多的cpu。

惰性策略：访问一个key的时候判断这个key是否达到过期时间了，过期了就删除。

定期删除：

隔一段时间，就随机抽取一些设置了过期时间的 key，检查其是否过期，如果过期就删除，
定期删除可能会导致很多过期 key 到了时间并没有被删除掉

Redis 会每秒进行十次过期扫描，过期扫描不会遍历容器中所有的 key，而是采用一种特殊策略

从容器中随机 20 个 key；
删除这 20 个 key 中已经过期的 key；
如果过期的 key 比率超过 1/4，那就重复步骤 1；

惰性删除：

当某个客户端试图访问key时，发现该key已超时会把此key从内存中删除

如果Redis是主从复制
- 从节点不会让key过期，而是主节点的key过期删除后，成为del命令传输到从节点进行删除
- 主库在 key 到期时，会在 AOF 文件里增加一条 del 指令，同步到所有的从库，从库通过执行这条 del 指令来删除过期的 key
- 指令同步是异步进行的，所以主库过期的 key 的 del 指令没有及时同步到从库的话，会出现主从数据的不一致，主库没有的数据在从库里还存在

Redis服务器实际使用的是惰性删除和定期删除两种策略
- 通过配合使用这两种删除策略，服务器可以很好地在合理使用CPU时间和避免浪费内存空间之间取得平衡。
- 问题
  - 如果定期删除漏掉了很多过期 key，然后你也没及时去查，也就没走惰性删除，此时会怎么样？
  - 如果大量过期 key 堆积在内存里，导致 redis 内存块耗尽了，就需要走内存淘汰机制

设计模式-单例创建

懒汉方式：就是所谓的懒加载，延迟创建对象
- 优点：前期不占据应用内存，用时创建
- 缺点: 初次创建对象有延迟
饿汉方式：提前创建好对象
- 优点：实现简单，使用时没延迟
- 缺点：不管有没使用，instance对象一直占着这段内存
如何选择：
- 如果对象不大，且创建不复杂，直接用饿汉的方式即可
- 其他情况则采用懒汉实现方式

2.惰性思想更新流量包

不用每天更新全部流量包，用的时候更新就可以了。

步骤

查询用户全部可用流量包
遍历用户可用流量包
- 判断是否更新-用日期判断（要么都更新过，要么都没更新，根据gmt_modified）
  - 没更新的流量包后加入【待更新集合】中
    - 增加【今天剩余可用总次数】
  - 已经更新的判断是否超过当天使用次数
    - 如果没超过则增加【今天剩余可用总次数】
    - 超过则忽略
更新用户今日流量包相关数据
扣减使用的某个流量包使用次数

  `id` bigint unsigned NOT NULL AUTO_INCREMENT,

  `day_limit` int DEFAULT NULL COMMENT '每天限制多少条，短链',

  `day_used` int DEFAULT NULL COMMENT '当天用了多少条，短链',

  `expired_date` date DEFAULT NULL COMMENT '过期日期',

  `gmt_modified` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,流量包更新时间

开发步骤：

1.trafficmanager层开发

	//查找可用的短链流量包(未过期),包括免费流量包

    List<TrafficDO> selectAvailableTraffics(Long accountNo);

     /* 给某个流量包增加使用次数,对某个trafficId的流量包增加usedTimes次，由trafficmapper和traffic.xml实现，xml里的思路

        update traffic set day_used = day_used + #{usedTimes}

        where id = #{trafficId} and account_no = #{accountNo}

        and (day_limit - day_used) >= #{usedTimes} limit 1 */

    int addDayUsedTimes(Long accountNo, Long trafficId, Integer usedTimes) ;  

/*恢复流量包使用当天次数,减去当天use次数

         update traffic set day_used = day_used - #{usedTimes}

         where id = #{trafficId} and account_no = #{accountNo}

         and (day_used - #{usedTimes}) >= 0 and date_format(gmt_modified,'%Y-%m-%d') = #{useDateStr} limit 1;

     */

    int releaseUsedTimes(Long accountNo, Long trafficId, Integer useTimes,String useDateStr);

     // 批量更新流量包使用次数为0

    int batchUpdateUsedTimes(Long accountNo, List<Long> unUpdatedTrafficIds);

2.创建UseTrafficVO信息类

    /**

     * 天剩余可用总次数 = 总次数 - 已用

     */

    private Integer dayTotalLeftTimes;

    /**

     * 当前使用的流量包

     */

    private TrafficDO currentTrafficDO;

    /**

     * 记录没过期，但是今天没更新的流量包id

     */

    private List<Long> unUpdatedTrafficIds;

3.reduce方法中调用上述方法以及类

   /**

     * * 查询用户全部可用流量包

     * * 遍历用户可用流量包

     *   * 判断是否更新-用日期判断

     *     * 没更新的流量包后加入【待更新集合】中

     *       * 增加【今天剩余可用总次数】

     *     * 已经更新的判断是否超过当天使用次数

     *       * 如果没超过则增加【今天剩余可用总次数】

     *       * 超过则忽略

     *

     * * 更新用户今日流量包相关数据

     * * 扣减使用的某个流量包使用次数

     * @param useTrafficRequest

     * @return

     */

    @Override

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRED)

    public JsonData reduce(UseTrafficRequest trafficRequest) {

        Long accountNo = trafficRequest.getAccountNo();

        //处理流量包，筛选出未更新流量包，当前使用的流量包

        UseTrafficVO useTrafficVO = processTrafficList(accountNo);

        log.info("今天可用总次数:{},当前使用流量包:{}",useTrafficVO.getDayTotalLeftTimes(),useTrafficVO.getCurrentTrafficDO());

        if(useTrafficVO.getCurrentTrafficDO() == null){

            return JsonData.buildResult(BizCodeEnum.TRAFFIC_REDUCE_FAIL);

        }

        log.info("待更新流量包列表:{}",useTrafficVO.getUnUpdatedTrafficIds());

		//没更新的话就更新

        if(useTrafficVO.getUnUpdatedTrafficIds().size()>0){

            //更新今日流量包

          trafficManager.batchUpdateUsedTimes(accountNo,useTrafficVO.getUnUpdatedTrafficIds());

        }

        //先更新，再扣减当前使用的流量包

        int rows = trafficManager.addDayUsedTimes(accountNo,useTrafficVO.getCurrentTrafficDO().getId(),1);

        TrafficTaskDO trafficTaskDO = TrafficTaskDO.builder().accountNo(accountNo).bizId(trafficRequest.getBizId())

                .useTimes(1).trafficId(useTrafficVO.getCurrentTrafficDO().getId())

                .lockState(TaskStateEnum.LOCK.name()).build();

        trafficTaskManager.add(trafficTaskDO);

        if(rows != 1){

            throw new BizException(BizCodeEnum.TRAFFIC_REDUCE_FAIL);

        }

        //往redis设置下总流量包次数，短链服务那边递减即可； 如果有新增流量包，则删除这个key

        long leftSeconds = TimeUtil.getRemainSecondsOneDay(new Date());

		//今天结束剩余的秒数

        String totalTrafficTimesKey = String.format(RedisKey.DAY_TOTAL_TRAFFIC,accountNo);

		//设置key-value key：lock:traffic:day_total:accountno值  value：总剩余次数-1，到今天结束过期

        redisTemplate.opsForValue().set(totalTrafficTimesKey,

                useTrafficVO.getDayTotalLeftTimes()-1,leftSeconds, TimeUnit.SECONDS);

        EventMessage trafficUseEventMessage = EventMessage.builder().accountNo(accountNo).bizId(trafficTaskDO.getId() + "")

                .eventMessageType(EventMessageType.TRAFFIC_USED.name()).build();

        //发送延迟消息，用于异常回滚

        rabbitTemplate.convertAndSend(rabbitMQConfig.getTrafficEventExchange(),

                rabbitMQConfig.getTrafficReleaseDelayRoutingKey(),trafficUseEventMessage);

        return JsonData.buildSuccess();

    }

/*

1.查找未过期流量列表（不一定可用，可能超过次数），免费的有free_init，付费的看过期日期

2.遍历全部流量包，如果今天流量包已经更新，那么dayLeftTimes天剩余可用总次数 = 总次数 - 已用，选取当次使用的流量包

如果流量包没有更新，那么更新流量包，并且把没更新的流量包的id列表记录下来

3.返回的UseTrafficVO，有dayTotalLeftTimes，UseTrafficDO信息，以及要是未更新的话，unUpdatedTrafficIds

*/

  private UseTrafficVO processTrafficList(Long accountNo) {

        //查找未过期流量列表（不一定可用，可能超过次数），免费的有free_init，付费的看过期日期

        List<TrafficDO> list = trafficManager.selectAvailableTraffics(accountNo);

        if(list == null || list.size()==0){ throw  new BizException(BizCodeEnum.TRAFFIC_EXCEPTION); }

        //天剩余可用总次数

        Integer dayTotalLeftTimes = 0;

        //当前使用

        TrafficDO currentTrafficDO = null;

        //没过期，但是今天没更新的流量包id列表

        List<Long> unUpdatedTrafficIds = new ArrayList<>();

        //今天日期

        String todayStr = TimeUtil.format(new Date(),"yyyy-MM-dd");

        //遍历全部流量包

        for(TrafficDO trafficDO : list){

            String trafficUpdateDate = TimeUtil.format(trafficDO.getGmtModified(),"yyyy-MM-dd");

            if(todayStr.equalsIgnoreCase(trafficUpdateDate)){

                //已经更新  天剩余可用总次数 = 总次数 - 已用

                int dayLeftTimes = trafficDO.getDayLimit() - trafficDO.getDayUsed();

                dayTotalLeftTimes = dayTotalLeftTimes+dayLeftTimes;

                //选取当次使用流量包

                if(dayLeftTimes>0 && currentTrafficDO==null){

                    currentTrafficDO = trafficDO;

                }

            }else {

                //未更新

                dayTotalLeftTimes = dayTotalLeftTimes + trafficDO.getDayLimit();

                //记录未更新的流量包

                unUpdatedTrafficIds.add(trafficDO.getId());

                //选取当次使用流量包

                if(currentTrafficDO == null){

                    currentTrafficDO = trafficDO;

                }

            }

        }

        UseTrafficVO useTrafficVO = new UseTrafficVO(dayTotalLeftTimes,currentTrafficDO,unUpdatedTrafficIds);

        return useTrafficVO;

    }

秒杀系统

使用sential的方式进行分布式限流

秒杀系统

1.除了这些服务的保护措施，秒杀的核心流程，比如有库存服务，订单服务，秒杀服务。首先要使用分布式调度把秒杀活动的商品信息加载到秒杀服务中，并且提前一天把商品配置到redis中里去。比如用户要参与秒杀活动，这时候需要在redis里预扣减库存，如果直接操作数据库的话肯定崩了。如果redis中扣减库存成功了，发送mq消息，如果失败的话直接向用户返回秒杀失败。发送mq消息后，订单服务进行消息消费，这边才做RPC调用，扣减库存做订单和商品库存的持久化。然后这个过程前端就是支付界面一个loading的加载界面，几秒后刷新然后调用支付服务，进入支付链路，这样就完成了一次秒杀。

https://www.cnblogs.com/zeenzhou/p/14668696.html 分段锁库存

假如你现在iphone有1000个库存，那么你完全可以给拆成20个库存段，要是你愿意，可以在数据库的表里建20个库存字段，比如stock_01，stock_02，类似这样的，也可以在redis之类的地方放20个库存key。
简单来说，把你的1000件库存给他拆开，每个库存段是50件库存，比如stock_01对应50件库存，stock_02对应50件库存。
接着，每秒1000个请求过来了，好！此时其实可以是自己写一个简单的随机算法，每个请求都是随机在20个分段库存里，选择一个进行加锁。
同时可以有最多20个下单请求一起执行，每个下单请求锁了一个库存分段，然后在业务逻辑里面，就对数据库或者是Redis中的那个分段库存进行操作即可，包括查库存 -> 判断库存是否充足 -> 扣减库存。
如果不用分段库存这1000个库存用分布式锁每个用户进行查库存 -> 判断库存是否充足 -> 扣减库存，如果每次要20ms，那么1000个库存需要20s。如果把1000个库存分为20个库分段加锁，这相当于一个20毫秒，可以并发处理掉20个下单请求，那么1000个库存需要1s就可以进行处理掉。
一旦对某个数据做了分段处理之后，有一个坑大家一定要注意：就是如果某个下单请求，咔嚓加锁，然后发现这个分段库存里的库存不足了，此时咋办？
这时你得自动释放锁，然后立马换下一个分段库存，再次尝试加锁后尝试处理。这个过程一定要实现。

短链服务和流量包服务

1.如果创建短链，然后rpc调用扣减流量包服务，50ms*3，然后再发送mq消息双写，这样性能会很差，所以参照秒杀来实现这部分功能。

预扣减

首先创建creatshortlink创建短链，然后这时候在redis里对流量包进行预扣减，redis中存储的是用户今日剩余的流量包，如果流量包次数不足，那么直接返回流量包不足。如果流量包足够，那么发送mq消息，经过b端，c端消费，创建两种类型的短链。然后消费者种rpc调用扣减流量包的请求，扣减流量包中用到的是查询全部流量包，更新今日流量包，还有再扣减使用的流量包。然后把流量包的过期时间，次数放到redis

分布式事务问题

短链服务创建短链成功的时候会在短链数据库中创建短链相关的数据，同时在创建短链的时候，也会调用流量包服务对相应的流量包使用次数进行扣减。这时候如果短链创建失败了，但是流量包次数又扣减了，就会有短链库和流量包库数据不一致的分布式事务问题。

我采用的解决方案是在流量包服务扣减库存前保存一个task任务，记录扣减的流量包，扣减更新流量包和保存task任务也要放在同一个事务下面。如果这时候流量包扣减了，但是创建短链失败了，可以通过分布式调度轮训task表，检查超时未完成的任务，比如超过十分钟记录，这个记录未更新，就要检查短链是否存在，如果不存在，那么就恢复流量包次数进行回滚。

详细说明：task表中有accuntno，traffic_id 流量包id，use_time,使用次数，lock_state 用于记录状态，有lock，finish，cancel状态，biz_id 短链码。在扣减流量包的时候先会往task表中插入相应的数据，比如流量包的id，最新的使用次数，然后锁定状态，然后会发送一个mq消息，可以延时5分钟，或者十分钟，然后时间到了之后流量包消费者执行相应的逻辑：首先

检查task表中的数据是否存在

如果存在就去检查短链是否成功

如果不成功，则恢复流量包的使用次数

删除task (也可以更新task状态，定时删除就行)

所以，这样就保证了短链库和流量包库最终都能成功创建以及使用次数扣减，或者全部失败。也就实现了数据的最终一致性。

总结

海量数据下采用惰性策略更新维护流量包

因为创建短链会消耗流量包中的次数，流量包可创建短链的次数需要每天更新，比如每天5次，所以流量包的使用次数需要更新维护。

关于更新维护的方案最容易想到的就是，定时任务-xxljob，可以每天用xxl-job调度中心定时执行账号微服务中的流量包更新服务

但是在海量数据下就会有问题，如果用户量大，会有更新延迟。比如每天0点定时更新，数据量一多，肯定会有更新延迟。所以就采用惰性思想来更新流量包，这里也借鉴了redis淘汰过期key的思想。不用每天更新全部流量包，只需要在用流量包的时候更新就可以了。比如创建短链，然后这个流量包是三天前用的还剩3次，但是每天可以用5次，这时候就可以在流量包扣减之前，把流量包的使用次数先更新到5次，然后再进行扣减。

高并发使用预扣减机制进行流量包的扣减

createShortLink链路

在ShortLinkController中的createShortLink（/api/link/v1/add路径下）的方法，调用shortLinkService.createShortLink(request)

然后执行以下逻辑：

1.预扣减：使用lua脚本检查key对应的value，如果存在就递减，返回剩余次数。如果不存在，那么返回0

2.次数足够，那么给url增加雪花前缀，设置EventMessage，rabbitmq给两个队列short_link.add.mapping.queue，short_link.add.link.queue发送消息

两个消费者监听到消息后，设置消息的属性（B端还是C端），然后执行shortLinkService.handleAddShortLink(eventMessage); 来进行对B端和C端的短链信息表的冗余双写，往商家短链数据库和用户短链数据库写两份数据。这个过程是通过rabbitmq实现的，如果满足创建短链的条件，那么发送一个带有指定路由key的消息，然后消息进入交换机，再进入两个队列，一个是要在用户数据库创建短链信息表的队列，另一个是要在商家短链信息表创建信息的队列。最后在编写两个相应的消费者监听这两个队列的消息，来执行向数据库双写的逻辑。

1.首先做合法性校验，判断域名，组名是否合法

2.生成短链码

3.然后用redis加lua脚本加锁

//先判断是否有，如没这个key，则设置key-value，配置过期时间，加锁成功

//如果有这个key,判断value是否是同个账号，是同个账号则返回加锁成功

//如果不是同个账号则加锁失败

String script = "if redis.call('EXISTS',KEYS[1])==0 then redis.call('set',KEYS[1],ARGV[1]); redis.call('expire',KEYS[1],ARGV[2]); return 1;" +

                " elseif redis.call('get',KEYS[1]) == ARGV[1] then return 2;" +

                " else return 0; end;";

加锁成功后B端和C端创建短链

1.B端

然后判断流量包次数是否足够reduceTraffic，这里通过feign调用了流量包服务中的trafficFeignService.useTraffic(request)，在创建的时候需要判断流量包的次数是否足够创建短链，如果不足就返回流量包次数不足。流量包次数是每天更新的，更新采用惰性策略来维护流量包的可用次数，更新后再进行流量包次数的扣减更新。因为这个短链的创建过程涉及到了对短链数据库和流量包数据库的操作，如果短链数创建失败，但是流量包次数还是扣减了，那么就造成了这两个数据库不一致的分布式事务问题，所以采用了本地task表加延时队列来实现最终一致性。最后创建短链成功

在短链数据冗余双写的过程也确保了最终一致性，比如一个创建短链的消息，分别到了B端的消费者和C端的消费者，如果C端的消费者消费成功，往用户短链数据库插入了数据，但是C端的消费者消费失败，商家的短链数据库没有数据，那么这次冗余双写只写进去了一个数据库。不过这时候消费失败的那个消息还是在消息队列里面，这个消息重试到一定次数就可以通过异常交换机通过告警服务让开发人员排查原因，解决问题之后确保数据库的一致。这个过程不能实现强一致性，但还是保证了最终的一致。常规情况下，这种情况很少，都会消费成功，但是以防万一也预留了解决方案。

大数据

对短链进行访客分析，需要知道PV、UV，新老访客、地理位置等等。

数据仓库 Data Warehouse
- 为了便于多维分析和多角度展现，将其数据按特定的模式进行存储而建立的数据库，数据仓库中的数据是细节的，集成的，面向主题的，是以 OLAP系统为分析目的，
- 是存储和管理一个或多个主题数据的集合，支持管理决策分析，有针对性抽取的结构化历史数据，能够生成各类报表
- 将来自不同来源的结构化数据聚合起来，用于业务智能领域的比较和分析（BI商业智能），数据仓库是包含多种数据的存储库
- 数据仓库有两个局限
  - 一是只可以解决预先想到的问题, 需要提前建模
  - 二是数据已经被多次处理过，无法看见其最初状态
数据湖 Data Lake
- 存储任何形式（包括结构化和非结构化）和任何格式（包括文本、音频、视频和图像）的原始数据
- 数据不需要提前进行定义，在准备使用数据时再定义，提高了最高的灵活性与可扩展性
- 适合使用机器学习和深度学习进行使用，比如数据挖掘和数据分析，以及提取非结构化数据
- 一个新的概念，但落地还很多问题需要解决
总结：在企业中两者的作用是互补的，不能认为数据湖的出现是为了取代数据仓库
通用微服务+数据仓库详细链路
- 链路存在重叠，意思是多种方式都是可以的实现
- 短链平台的数据是其中一个链路

1.上面是微服务，下面是大数据链路。用户在浏览器点击、下单、停留数据，可以通过微服务获取。前端也可以通过埋点日志搜集用户的行为数据，然后推送到消息队列比如kafka，然后再进行数据处理flink，处理原始数据，然后分层处理ODS,DWD,DWM,DWS。数据分层处理完之后，可以通过数据可视化查看相应的数据，

数据采集传输:Flume、Kafka、Canal、Maxwell、Sqoop、Logstash，DataX
数据存储:MySql、ClickHouse、HDFS、HBase、Redis
数据计算:Hive、Spark、Flink、Storm
数据查询:Presto、Kylin、Druid
数据可视化:Echarts、Superset、DataV

数据仓库分层：

数据埋点方式：

代码埋点
- 编写埋点代码，通过代码进行控制，前端、后端、App客户端代码，需要埋点的逻辑通过sdk函数调用，上报数据
- 比如前端点击一个元素，都有click事件，可以通过这个统计相应的，后端也可以通过aop埋点
可视化埋点
就是用一个采集的SDK，只要把这个网址输入进去，就能识别出来有哪些元素可以进行采集代码。
- 通用采集SDK，项目只需要引入埋点采集sdk。分析人员通过分析平台进行操作，对可交互的页面元素（如：图片、按钮、链接等）直接在界面上进行操作实现数据埋点，下发采集代码生效的埋点方式。
- 所见即所得，使用者只需在其可视化埋点页面上，点击想要监测的元素，编辑名字编号等，埋点就完成了
- 缺点
  - 存在滞后性，每次调整埋点后需要应用重新发版才可以看到数据，也可以通过配置中心动态下发解决
  - 相对生硬，满足不了全部数据采集，比如编码规范不统一、无法定位元素等，或者需要调用后台接口的数据等
    - 比如看小滴课堂的一个视频，点击播放一个视频，交互行为就是一个播放，但播放的背后还想知道这个视频的名字、类别、作者、评级等信息就获取不了

数据日志采集

需求：

控制台输出访问日志，方便测试
业务数据实际输出到kafka
log4j、logback、self4j 之间有啥关系
- SLF4J(Simple logging Facade for Java) 门面设计模式 |外观设计模式
  - 把不同的日志系统的实现进行了具体的抽象化，提供统一的日志使用接口
  - 具体的日志系统就有log4j,logback等；
  - logback也是log4j的作者完成的，有更好的特性，可以取代log4j的一个日志框架, 是slf4j的原生实现
  - log4j、logback可以单独的使用，也可以绑定slf4j一起使用
- 编码规范建议不直接用log4j、logback的API，应该用self4j, 日后更换框架所带来的成本就很低
Logback知识点回顾
- appender是记录日志的方式 console
- logger是配置某个包或者类采用哪几个appender记录日志，比如控制台/文件；
- root是默认配置的输出日志，除了logger自定义配置外的其他类输出日志的方式，也是有级别和多个appender

<?xml version="1.0" encoding="UTF-8"?>

<configuration>

    <property name="LOG_HOME" value="./data/logs/link" />

    <!--采用打印到控制台，记录日志的方式-->

    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">

    </appender>

    <!-- 采用保存到日志文件 记录日志的方式-->

    <appender name="rollingFile" class="ch.qos.logback.core.rolling.RollingFileAppender">

        <file>${LOG_HOME}/link.log</file>

    </appender>

    <!-- 指定某个类单独打印日志 -->

    <logger name="net.xdclass.service.impl.LogServiceImpl"

            level="INFO" additivity="false">

        <appender-ref ref="rollingFile" />

        <appender-ref ref="console" />

    </logger>

    <root level="info" additivity="false">

        <appender-ref ref="console" />

    </root>

</configuration>

具体开发步骤：

1.在linkservice下的resource里创建logback.xml文件

2.创建LogService以及实现类

3.public void recordShortLinkLog(HttpServletRequest request, String shortLinkCode, Long accountNo) 方法

4.在LinkApiController的短链路径里使用这个方法来打印日志。在访问的时候，可以获取ip，以及全部的http请求头，在请求头中获取一些有用信息，比如user-agent，referrence，把这些信息可以通过控制台打印出来。然后把这些信息发送到kafka。（kafka需要创建topic）

Flink当中使用kafka Consumer

Flink处理

flink步骤

在通过logservice中将信息发送到kafka的ods_link_visit_topic中，也就是说，用户点击一个短链，在controller层写入了一个日志服务，记录用户的ip，还有http请求头等信息，然后把这些信息封装好发送到kafka的topic中，然后在这个消息队列中的数据作为flink中的初始数据来源来作下一步的处理，包括数据仓库分层，数据可视化处理等等。

DWD层

将之前通过logservice中获取的信息发送到kafka的ods_link_visit_topic中，将这个topic（kafka的consumer）作为flink的source数据来源，这里用到了FlinkKafkaConsumer，因为kafka可以作为Flink的Source和Sink来使用，这个api就可以作为一个连接器，让kafka中的topic里的数据作为flink的souce和sink（数据的输入和输出）。

再定义另一个topic作为sink dwd_link_visit_topic

//定义source topic

public static final String SOURCE_TOPIC = "ods_link_visit_topic";

//定义sink topic

public static final String SINK_TOPIC = "dwd_link_visit_topic";

//定义消费者组

public static final String GROUP_ID = "dwd_short_link_group";

新老访客：

将原始数据层（ODS）读取数据，处理后存储到DWD层，需要需要识别标记出短链的新老访客。这个过程就是ETL的过程，就是数据仓库中的，抽取，转换，加载的缩写。将业务系统的数据经过抽取、清洗转换之后加载到数据仓库的过程，目的是将企业中的分散、零乱、标准不统一的数据整合到一起，为企业的决策提供分析依据。

具体实现：

通过设备唯一标识，服务端进行【天维度】状态存储，标记新老访客
详情
- 需要生成唯一设备标识
- 利用Flink的状态存储 ValueState

唯一设备标识：要通过用户访问得到的信息（比如ip地址，http的user agent，或者cookie等），生成一个唯一标识

这里使用的就是ip+useragent（操作系统、浏览器类型等信息）

这里用到flatmap

深度解析FlinkflatMap算子的自定义方法

flatMap()

对于map()来说，实现MapFunction也只是支持一对一的转换。那么有时候你需要处理一个输入元素，但是要输出一个或者多个输出元素的时候，就可以用到flatMap()。

     //数据补齐

        SingleOutputStreamOperator<JSONObject> jsonDS = ds.flatMap(new FlatMapFunction<String, JSONObject>() {

            @Override

            public void flatMap(String value, Collector<JSONObject> out) throws Exception {

                JSONObject jsonObject = JSON.parseObject(value);

                //生成设备唯一id

                String udid = getDeviceId(jsonObject);

                jsonObject.put("udid",udid);

                String referer = getReferer(jsonObject);

                jsonObject.put("referer",referer);

                out.collect(jsonObject);

            }

        });

 /**

     * 生成设备唯一id

     * @param jsonObject

     * @return

     */

    public static String getDeviceId(JSONObject jsonObject){

        Map<String,String> map = new TreeMap<>();

        try {

            map.put("ip",jsonObject.getString("ip"));

            map.put("event",jsonObject.getString("event"));

            map.put("bizId",jsonObject.getString("bizId"));

            String userAgent = jsonObject.getJSONObject("data").getString("user-agent");

            map.put("userAgent",userAgent);

            String deviceId = DeviceUtil.geneWebUniqueDeviceId(map);

            return deviceId;

        }catch (Exception e){

            log.error("生成唯一deviceid异常:{}",jsonObject);

            return null;

        }

    }

来源访问：

比如一个短链可以投放到不同的网站，这时候就可以进行来源访问来查看哪个网站跳转的最多，然后就可以在来源最多的网站加大推广。

    /**

     * 提取referer

     * HTTP Referer是header的一部分，当浏览器向web服务器发送请求的时候，一般会带上Referer，

     * 告诉服务器该网页是从哪个页面链接过来的，服务器因此可以获得一些信息用于处理。

     * @param jsonObject

     * @return

     */

    public static String getReferer(JSONObject jsonObject){

        JSONObject dataJsonObj = jsonObject.getJSONObject("data");

        if(dataJsonObj.containsKey("referer")){

            String referer = dataJsonObj.getString("referer");

            if(StringUtils.isNotBlank(referer)){

                try {

                    URL url = new URL(referer);

                    return url.getHost();

                }catch (MalformedURLException e) {

                    log.error("提取referer失败:{}",e);

                }

            }

        }

        return "";

    }

流Stream与收集器Collector

分组.同个终端设备同个组,uuid作为key来进行分组,在电商中处理也很常见，比如对某个商品名称进行keyby分类，然后把该组内的商品价格进行相加，就可以统计出该商品类别的销售额。

分组完之后，对同一个设备id作为key分组，

   //分组.同个终端设备同个组

        KeyedStream<JSONObject, String> keyedStream = jsonDS.keyBy(new KeySelector<JSONObject, String>() {

            @Override

            public String getKey(JSONObject value) throws Exception {

                return value.getString("udid");

            }

        });

        //识别 richMap open函数，做状态存储的初始化

        SingleOutputStreamOperator<String> jsonDSWithVisitorState = keyedStream.map(new VistorMapFunction());

        jsonDSWithVisitorState.print("ods新老访客");

        //存储到dwd

        FlinkKafkaProducer<String> kafkaProducer = KafkaUtil.getKafkaProducer(SINK_TOPIC);

        jsonDSWithVisitorState.addSink(kafkaProducer);

VistorMapFunction extends RichMapFunction<JSONObject,String>

@Slf4j

public class VistorMapFunction extends RichMapFunction<JSONObject,String> {

    //记录用户的udid访问

    private ValueState<String> newDayVisitorState;

    @Override

    public void open(Configuration parameters) throws Exception {

        //对状态做初始化

        newDayVisitorState = getRuntimeContext().getState(new ValueStateDescriptor<String>("newDayVisitorState",String.class));

    }

    @Override

    public String map(JSONObject value) throws Exception {

        //获取之前是否有访问日期

        String beforeDateState = newDayVisitorState.value();

        //获取当前访问时间戳

        Long ts = value.getLong("ts");

        String currentDateStr = TimeUtil.format(ts);

        //判断日期是否为空进行新老访客识别

        if(StringUtils.isNotBlank(beforeDateState)){

            if(beforeDateState.equalsIgnoreCase(currentDateStr)){

                //一样则是老访客

                value.put("is_new",0);

                log.info("老访客:{}",currentDateStr);

            }else {

                //时间不一样，则是新用户,标记1，老访客标记0

                value.put("is_new",1);

                newDayVisitorState.update(currentDateStr);

                log.info("新访客:{}",currentDateStr);

            }

        }else {

            //如果状态为空，则是新用户,标记1，老访客标记0

            value.put("is_new",1);

            newDayVisitorState.update(currentDateStr);

            log.info("新访客:{}",currentDateStr);

        }

        return value.toJSONString();

    }

}

总结：

{"ip":"141.123.11.31","ts":1646145133665,"event":"SHORT_LINK_TYPE","udid":null,"bizId":"026m8O3a","data":{"referer":null,"accountNo":"693100647796441088","user-agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.109 Safari/537.36"}}

通过对ODS层原始数据的处理，这里的原始数据就是用户在访问的时候，通过在controller层埋点，发送json数据到kafka的topic中，这些json数据有ip，useragent,用户的基本信息等等，对这些数据进行抽取，转换，加载，可以抽取、转化出用户的唯一标识用来识别新老访客，referrece用来统计访问来源。这里的转换就用到了flink中常用的算子flatmap，他能把一个输入元素转换为一个或者多个输出。

这里用可以自己定义输入和输出的类型，定义的时候只声明输入和输出泛型的类型就可以了。在这里我们的输入类型就是String类型，输出就是转化后的提取了用户唯一设备标识和referce来源的数据，以及还有其他一些基本信息.然后对同一分组下的uuid进行新老访问的逻辑处理，初始数据中有时间戳，这里的新老用户判断是以天维度的，如果之前访问的日期不存在或者和现在不是同一天，那么就是新用户，反之就老用户。新老用户用is_new字段标识，如果是老用户用0来标识，新用户，用1来标识。

所以这时候就对原始数据进行了抽取，转化，加入了uuid设备唯一标识，以及新老用户的信息字段，然后将这个数据作为sink保存到dwd层。dwd_link_visit_topic

DWM层

ODS和DWD和业务关系不大，DWM、DWS和业务关系就大

什么是宽表和窄表
- 宽表（明细表）
  - 简单讲字段比较多的数据库表，通常是指业务主题相关的指标、维度、属性关联在一起的一张数据库表
  - 把不同的内容都放在同一张表存储，宽表不符合三范式的模型设计规范
  - 尽量满足多维，多度量，遵循维度建模的原则
  - 缺点：数据的大量冗余
  - 优点：减少表关联数量，查询性能的提高，空间换时间
- 窄表
  - 严格按照数据库设计三范式，尽量减少数据冗余
  - 缺点：做数据分析查询OLAP时，需要大量关联多个表，性能下降
  - 优点：存储省空间，大量数据只存储某个表

业务需求

需要得出短链访问的终端设备分布情况，做出【宽表】
- 浏览器类型分布 Chrome
- 操作系统分布 Android
- 设备类型分布 Mobile、Computer
- 设备生产厂商 GOOGLE、APPLE
- 系统版本 Android 10、Intel Mac OS X 10_15_7

因为我们需要得出短链访问的终端设备分布情况，包括浏览器类型，操作系统，设备厂商，系统版本等等，这里通过一个用来解析 User-Agent 字符串的 Java 类库UserAgentUtils识别浏览器名字，浏览器组，浏览器类型，浏览器版本...

步骤

    //2、格式装换，补齐设备信息

        SingleOutputStreamOperator<ShortLinkWideDO> deviceWideDS = ds.map(new DeviceMapFunction());

        deviceWideDS.print("设备信息宽表补齐");

        //3、补齐地理位置信息

        //SingleOutputStreamOperator<String> shortLinkWideDS  = deviceWideDS.map(new LocationMapFunction());

        SingleOutputStreamOperator<String> shortLinkWideDS = AsyncDataStream.unorderedWait(deviceWideDS, new AsyncLocationRequestFunction(), 1000, TimeUnit.MILLISECONDS, 200);

        shortLinkWideDS.print("地理位置信息宽表补齐");

        FlinkKafkaProducer<String> kafkaProducer = KafkaUtil.getKafkaProducer(SINK_TOPIC);

        //4、将sink写到dwm层，kafka存储

        shortLinkWideDS.addSink(kafkaProducer);

        env.execute();

1.补齐设备信息

public class DeviceMapFunction implements MapFunction<String, ShortLinkWideDO> {

    @Override

    public ShortLinkWideDO map(String value) throws Exception {

        //还原json

        JSONObject jsonObject = JSON.parseObject(value);

        String userAgent = jsonObject.getJSONObject("data").getString("user-agent");

        DeviceInfoDO deviceInfoDO = DeviceUtil.getDeviceInfo(userAgent);

        String udid = jsonObject.getString("udid");

        deviceInfoDO.setUdid(udid);

        //配置短链基本信息宽表

        ShortLinkWideDO shortLinkWideDO = ShortLinkWideDO.builder()

                //短链访问基本信息

                .visitTime(jsonObject.getLong("ts"))

                .accountNo(jsonObject.getJSONObject("data").getLong("accountNo"))

                .code(jsonObject.getString("bizId"))

                .referer(jsonObject.getString("referer"))

                .isNew(jsonObject.getInteger("is_new"))

                .ip(jsonObject.getString("ip"))

                //设备信息补齐

                .browserName(deviceInfoDO.getBrowserName())

                .os(deviceInfoDO.getOs())

                .osVersion(deviceInfoDO.getOsVersion())

                .deviceType(deviceInfoDO.getDeviceType())

                .deviceManufacturer(deviceInfoDO.getDeviceManufacturer())

                .udid(deviceInfoDO.getUdid())

                .build();

        return shortLinkWideDO;

    }

}

2.补齐地理位置

这里的地理信息是根据ip信息转换为地理位置信息，一般有两种做法，可以用离线的一些ip库或者用在线的api比如百度、高德支持的api，这里用的是调用api接口，返回请求参数中指定上网ip以及你的密钥，就能得到大致地理位置信息。这里数据格式的转化就是将宽表的do类转为string类型的地理位置信息，一对一的转化用map函数，又因为这里需要发送http请求，所以要建议http连接池进行优化。所以用richmap函数来进行优化， richmap中有open和close，可以在open方法中来定义http连接池，colse方法来关闭，这两个就是生命周期的开始和结束。中间部分的逻辑就是把数据do类中的ip执行拿出来，然后向第三方服务商规定的url发送http请求，这里包括了ip地址以及用户的key（比如高德地图的用户key，用来验证你有没有使用这个功能的权限），然后把相应的地理位置信息插入到do类，转为string类型输出。

优化：

使用了http连接池，但是这是同步阻塞的过程，如果在海量请求下，性能就会很低。所以需要异步处理，在spring中可以用Async注解来实现异步。在flink中，很多情况下会和外部表关联去补充更多维度属性信息，需要大量外部的交互，如HTTP网络、Redis、Mysql数据库、Hbase等进行查询。

默认Flink里面用 MapFunction进行对象关联，只能用同步方式去进行IO调用，需要等请求完成才进行发下一个请求，这种等待占了函数时间的绝大部分，虽然可以设置高并行度，但是高并行度会占用更多的资源，flink引入了异步io，单个并行函数可以同时处理多个请求并同时接收响应，哪个请求先返回就先处理，在连续的请求的时候不需要阻塞式等待。

3.PV和UV

UV（Uniqued Visitor）
- 独立访客就是独立IP访客（Unique Visitor）,访问网站的一台电脑客户端为一个访客，在 00:00-24:00内相同的客户端只被计算一次。
- 记录独立访客数的时间标准一般可为一天，一个月，一般不计算年UV数
PV（Page View）
- 页面浏览量或点击量，用户每次刷新即被计算一次。指某站点总共有被浏览多少个页面，它是重复累计的，同一个页面被重复浏览也被计入PV。

UV是统计日活的UV，只要访问过短链的就算日活跃用户

大体思路是1.需要知道用户的唯一ID2.需要知道访问时间3.如果是同一天访问的就可以去重

对用户设备的唯一标识分组，利用keyby函数，将用户的唯一标识作为key，将所有设备标识相同的用户分为一组，然后进行去重操作。

flink中有无状态流和有状态流，就比如之前的map算子，每有一个用户点击短链，就生成相应的信息，这是无状态流。而这里的pv，uv，会根据每条输入进行更新，同一个用户一天内多次点击uv不会变，pv会增加。

所以对设备标识分组后的去重操作，用到了key的状态。Flink为每个key维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中。

具体实现是可以设置键状态的过期时间，这里统计天活跃用户，就把过期时间设为一天。所以把访问时间设置过期时间为一天，如果过期这个值就会被清理变成null，每次判断只要这个值不为空，且这次时间与之前的日期值相等，就不计入uv值，做到了去重。

Flink的状态管理

总结

将DWD层的数据作为DWM层FLINK的source输入，进行数据的格式转换，这里输入的字段是String类型的DWD层的数据，转换为宽表的DO类，将解析出来的user-agent里的具体信息比如浏览器，操作系统都提取出来，还要加上地理位置的字段（这是根据ip）为了后续更好的注入宽表。宽表字段有短链业务本身信息，比如短链码，账号标识码，访问来源，还有设备相关的字段，浏览器、操作系统版本等等，以及地理位置，省份城市运营商等等。最终将这些信息构建到宽表的do对象。这里的地理信息是根据ip信息转换为地理位置信息，一般有两种做法，可以用离线的一些ip库或者用在线的api比如百度、高德支持的api，这里用的是调用api接口，返回请求参数中指定上网ip以及你的密钥，就能得到大致地理位置信息。将所有信息插入宽表。

DWM层还有一个处理，就是统计pv，uv。将上面的数据保存到kafka的topic中，作为这一次处理的中间输入，增加pv和uv的统计

UV是统计日活的UV，只要访问过短链的就算日活跃用户

大体思路是1.需要知道用户的唯一ID2.需要知道访问时间3.如果是同一天访问的就可以去重

对用户设备的唯一标识分组，利用keyby函数，将用户的唯一标识作为key，将所有设备标识相同的用户分为一组，然后进行去重操作。

所以对设备标识分组后的去重操作，用到了key的状态。Flink为每个key维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中。

具体实现是可以设置键状态的过期时间，这里统计天活跃用户，就把过期时间设为一天。所以把访问时间设置过期时间为一天，如果过期这个值就会被清理变成null，每次判断只要这个值不为空，且这次时间与之前的日期值相等，就不计入uv值，做到了去重，这里还没有做pv的统计，只做了判断是否是独立访客uv。因为这里对数据进行了两次操作，第一次将DWD层的数据提取出了一些用户信息，以及user=agent里的浏览器，以及增加了地理位置，第二次又判断了是不是uv。所以这里就有两个数据流，便于下一步在DWS层继续处理。

DWS层

数据分析中最重要、最基础的2个概念：维度Dimensions 和度量Measures
- 度量是数据表中的数值数据，维度是类别数据
  - 例子：各个城市访问的UV数量
- 一个数据指标一般由一种或多种维度加上一种度量组成
需求统计点
- DWS层数据是做啥的?
  - 度量：PV、UV
  - 维度 : 新老用户、地区信息、设备信息等
  - DWS层数据处理后-存储到ClickHouse、Redis、Mysql、ElasticSearch等都可以
  - ADS层需要使用的数据就从上述的DWS层进行读取，主要是根据各种报表及可视化来生成统计数据。

数据分层

数据分层	分层描述	数据生成计算工具	存储
ODS	原生数据，短链访问基本信息	SpringBoot生成	Kafka
DWD	对 ODS 层做数据清洗和规范化，新老访客标记等	Flink	Kafka
DWM	对DWD数据进一步加工补齐数据，独立访客统计，操作系统/ip/城市，做宽表	Flink	kafka
DWS	对DWM进行处理，多流合并,分组\|聚合\|开窗\|统计，形成主题宽表，轻量聚合	Flink	ClickHouse
ADS	从ClickHouse中读取数据，根据需求进行筛选聚合，可视化展示	ClickHouseSql	web可视化展示

统计指标
- PV -》从DWM层 dwm_link_visit_topic统计
- UV-》从DWM层 dwm_unique_visitor_topic统计

因为DWM层对数据进行了两次操作，第一次将DWD层的数据提取出了一些用户信息，以及user=agent里的浏览器，以及增加了地理位置，第二次又判断了是不是uv。所以这里就有两个数据流，便于下一步在DWS层继续处理。

新建一个另一个宽表的do类，这里是对DWM进行处理，形成进一步的主题宽表。

使用flink的union算子进行多流合并，数据会按照先入先出且不去重的方式合并。

步骤

1.dwm里的两个sink输出流存储在kafka中，可以直接获取这两个个数据流，然后再做数据对齐，因为DWM层设计的宽表的字段是没有pv和uv的，新增的宽表添加了一些新的字段，将已有的字段直接填入新的宽表，这两条数据流中，一条做pv统计，uv置为0，一条做uv统计，pv置为0.然后用union算子进行多流合并。

2.设置WaterMark Flink的WaterMark详解

3.设置多维度、多个字段分组，比如某个省的pv，uv，某个操作系统的pv，uv。还是用keyby来进行分组。

进行分组的时候把需要分组的字段存储在tuple元组里面，这时候可以根据元组里的字段来进行多维度分组。

（元组和列表list一样，都可能用于数据存储，包含多个数据；但是和列表不同的是：列表只能存储相同的数据类型，而元组不一样，它可以存储不同的数据类型，比如同时存储int、string、list等，并且可以根据需求无限扩展。）

6.然后用滚动窗口来对数据划分统计，这里开窗的方式用了tumbling，也就是无重叠数据的窗口，窗口时间设置为10秒，就是每隔10秒统计pv，uv数

7.聚合统计，通过flink的reduce算子，对滑动窗口中的数据（比如pv，uv）进行聚合累加，window算子之后的reduce，其实计算的是window窗口内的数据和，每次窗口触发的时候，才会输出一次结果。也就是每一个时间窗口（10秒）就把之前的pv，uv相加，就能得到实时的pv，uv了。

8.将所有字段输出保存到Clickhouse

。

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //1、获取多个数据流，dwm里的两个sink输出流当作输入

        FlinkKafkaConsumer<String> shortLinkSource = KafkaUtil.getKafkaConsumer(SHORT_LINK_SOURCE_TOPIC,SHORT_LINK_SOURCE_GROUP);

        DataStreamSource<String> shortLinkDS = env.addSource(shortLinkSource);

        FlinkKafkaConsumer<String> uniqueVisitorSource = KafkaUtil.getKafkaConsumer(UNIQUE_VISITOR_SOURCE_TOPIC, UNIQUE_VISITOR_SOURCE_GROUP);

        DataStreamSource<String> uniqueVisitorDS = env.addSource(uniqueVisitorSource);

        //2、结构转换 uniqueVisitorDS、shortLinkDS

        //这条做pv统计，uv必须为0

        SingleOutputStreamOperator<ShortLinkVisitStatsDO> shortLinkMapDS = shortLinkDS.map(new MapFunction<String, ShortLinkVisitStatsDO>() {

            @Override

            public ShortLinkVisitStatsDO map(String value) throws Exception {

                ShortLinkVisitStatsDO visitStatsDO = parseVisitStats(value);

                visitStatsDO.setPv(1L);

                visitStatsDO.setUv(0L);

                return visitStatsDO;

            }

        });

        //独立访客          //这条做uv统计，pv必须为0

        SingleOutputStreamOperator<ShortLinkVisitStatsDO> uniqueVisitorMapDS = uniqueVisitorDS.map(new MapFunction<String, ShortLinkVisitStatsDO>() {

            @Override

            public ShortLinkVisitStatsDO map(String value) throws Exception {

                ShortLinkVisitStatsDO visitStatsDO = parseVisitStats(value);

                visitStatsDO.setPv(0L);

                visitStatsDO.setUv(1L);

                return visitStatsDO;

            }

        });

        //3、多流合并（合并相同结构的流）

        DataStream<ShortLinkVisitStatsDO> unionDS = shortLinkMapDS.union(uniqueVisitorMapDS);

        //4、设置WaterMark

        SingleOutputStreamOperator<ShortLinkVisitStatsDO> watermarkDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy

                //指定允许乱序延迟最大3秒

                .<ShortLinkVisitStatsDO>forBoundedOutOfOrderness(Duration.ofSeconds(3))

                //指定事件时间列，毫秒

                .withTimestampAssigner((event, timestamp) -> event.getVisitTime()));

        //5、多维度、多个字段分组

        // code、referer、isNew

        // province、city、ip

        // browserName、os、deviceType

        KeyedStream<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>> keyedStream = watermarkDS.keyBy(new KeySelector<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>>() {

            @Override

            public Tuple9<String, String, Integer, String, String, String, String, String, String> getKey(ShortLinkVisitStatsDO obj) throws Exception {

                return Tuple9.of(obj.getCode(), obj.getReferer(), obj.getIsNew(),

                        obj.getProvince(), obj.getCity(), obj.getIp(),

                        obj.getBrowserName(), obj.getOs(), obj.getDeviceType());

            }

        });

        //6、开窗 10秒一次数据插入到 ck

        WindowedStream<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>, TimeWindow> windowedStream =

                keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        //7、聚合统计(补充统计起止时间)

        SingleOutputStreamOperator<Object> reduceDS = windowedStream.reduce(new ReduceFunction<ShortLinkVisitStatsDO>() {

            @Override

            public ShortLinkVisitStatsDO reduce(ShortLinkVisitStatsDO value1, ShortLinkVisitStatsDO value2) throws Exception {

                value1.setPv(value1.getPv() + value2.getPv());

                value1.setUv(value1.getUv() + value2.getUv());

                return value1;

            }

        }, new ProcessWindowFunction<ShortLinkVisitStatsDO, Object, Tuple9<String, String, Integer, String, String, String, String, String, String>, TimeWindow>() {

            @Override

            public void process(Tuple9<String, String, Integer, String, String, String, String, String, String> tuple,

                                Context context, Iterable<ShortLinkVisitStatsDO> elements, Collector<Object> out) throws Exception {

                for (ShortLinkVisitStatsDO visitStatsDO : elements) {

                    //窗口开始和结束时间

                    String startTime = TimeUtil.formatWithTime(context.window().getStart());

                    String endTime = TimeUtil.formatWithTime(context.window().getEnd());

                    visitStatsDO.setStartTime(startTime);

                    visitStatsDO.setEndTime(endTime);

                    out.collect(visitStatsDO);

                }

            }

        });

        reduceDS.print(">>>>>>");

        //8、输出Clickhouse

        //8、输出Clickhouse

        String sql = "insert into visit_stats values(?,?,?,? ,?,?,?,? ,?,?,?,? ,?,?,?)";

        reduceDS.addSink(MyClickHouseSink.getJdbcSink(sql));

        env.execute();

ADS层

clickHouse 与mysql等关系型数据库对比

从ClickHouse中读取数据，根据需求进行筛选聚合，可视化展示

根据web可视化报表统计需求，从ClickHouse聚合统计

简单来说就是编写sql语句从clickhouse中进行筛选统计

1.分页实时查看访问记录

需要规定访问时间，以及访问条数。比如近一个月的5000条次数，这个可以从controller层进行控制

  public Map<String, Object> pageVisitRecord(VisitRecordPageRequest request) {

        Long accountNo = LoginInterceptor.threadLocal.get().getAccountNo();

        Map<String, Object> data = new HashMap<>(16);

        String code = request.getCode();

        int page = request.getPage();

        int size = request.getSize();

        int count = visitStatsMapper.countTotal(code, accountNo);

        int from = (page - 1) * size;

        List<VisitStatsDO> list = visitStatsMapper.pageVisitRecord(code, accountNo, from, size);

        List<VisitStatsVO> visitStatsVOS = list.stream().map(obj -> beanProcess(obj)).collect(Collectors.toList());

        data.put("total", count);

        data.put("current_page", page);

        /**计算总页数*/

        int totalPage = 0;

        if (count % size == 0) {

            totalPage = count / size;

        } else {

            totalPage = count / size + 1;

        }

        data.put("total_page", totalPage);

        data.put("data", visitStatsVOS);

        return data;

    }

 <!--统计总条数-->

    <select id="countTotal" resultType="java.lang.Integer">

        select count(1) from visit_stats  where  account_no=#{accountNo} and code=#{code} limit 1000

    </select>

    <!--分页查找-->

    <select id="pageVisitRecord" resultMap="BaseResultMap">

        select

        <include refid="Base_Column_List"/>

        from visit_stats where account_no=#{accountNo} and code=#{code}

        order by ts desc limit #{from},#{size}

    </select>

2.时间范围内地区访问分布

@Override

public List<VisitStatsVO> queryRegionWithDay(RegionQueryRequest request) {

    Long accountNo = LoginInterceptor.threadLocal.get().getAccountNo();

    List<VisitStatsDO> list = visitStatsMapper.queryRegionVisitStatsWithDay(request.getCode(), accountNo, request.getStartTime(), request.getEndTime());

    List<VisitStatsVO> visitStatsVOS = list.stream().map(obj -> beanProcess(obj)).collect(Collectors.toList());

    return visitStatsVOS;

}

<!--时间范围内地区访问分布-城市级别，天级别-->

<select id="queryRegionVisitStatsWithDay" resultMap="BaseResultMap">

    select province,city,sum(pv) pv_count, sum(uv) uv_count,count( DISTINCT ip) ip_count from visit_stats

    where  account_no=#{accountNo} and code=#{code} and toYYYYMMDD(start_time) BETWEEN #{startTime} and #{endTime}

    group by province,city order by pv_count desc

</select>

3.天维度访问曲线图接口实战

@Data

public class VisitTrendQueryRequest {

    private String code;

    /**

     * 跨天、当天24小时、分钟级别

     */

    private String type;

    private String startTime;

    private String endTime;

}

/**

     * 查询访问趋势，支持多天查询，支持查询当天小时级别

     *

     * @param request

     * @return

     */

    @Override

    public List<VisitStatsDO> queryVisitTrend(VisitTrendQueryRequest request) {

        Long accountNo = LoginInterceptor.threadLocal.get().getAccountNo();

        String code = request.getCode();

        String type = request.getType();

        List<VisitStatsDO> list = null;

       if (DateTimeFieldEnum.DAY.name().equalsIgnoreCase(type)) {

            list = visitStatsMapper.queryVisitTrendWithMultiDay(code, accountNo, request.getStartTime(), request.getEndTime());

        }

                List<VisitStatsVO> visitStatsVOS = list.stream().map(obj -> beanProcess(obj)).collect(Collectors.toList());

        return visitStatsVOS;

    }

<!-- 多天内的访问曲线图,天基本 -->

<select id="queryVisitTrendWithMultiDay" resultMap="BaseResultMap">

        select toYYYYMMDD(start_time) date_time_str,sum(if(is_new='1', visit_stats.uv,0)) new_uv_count,

 sum(visit_stats.uv) uv_count, sum(pv) pv_count, count( DISTINCT ip) ip_count from visit_stats

 where   account_no=#{accountNo} and code=#{code} and toYYYYMMDD(start_time) BETWEEN #{startTime} and #{endTime} group by date_time_str ORDER BY date_time_str desc

</select>

select toYYYYMMDD(start_time) date_time_str,sum(if(is_new='1', visit_stats.uv,0)) new_uv_count,

 sum(visit_stats.uv) uv_count, sum(pv) pv_count, count( DISTINCT ip) ip_count from visit_stats

 where   account_no=693100647796441088 and code='026m8O3a' and toYYYYMMDD(start_time)

 BETWEEN '20220303' and '20220430' group by date_time_str ORDER BY date_time_str desc

4.访问来源Top10统计开发

@Data

public class BaseQueryRequest {

    private String code;

    private String startTime;

    private String endTime;

}

	/**

     * 高频访问来源

     *

     * @param request

     * @return

     */

    @Override

    public List<VisitStatsDO> queryFrequentReferer(BaseQueryRequest request) {

        Long accountNo = LoginInterceptor.threadLocal.get().getAccountNo();

        List<VisitStatsDO> list = visitStatsMapper.queryFrequentReferer(request.getCode(), accountNo, 10);

        List<VisitStatsVO> visitStatsVOS = list.stream().map(obj -> beanProcess(obj)).collect(Collectors.toList());

        return visitStatsDOS;

    }

    <!--高频referer查询 访问来源-->

    <select id="queryFrequentSource" resultMap="BaseResultMap">

select referer,sum(pv) pv_count from visit_stats where  account_no=#{accountNo} and code=#{code} and toYYYYMMDD(start_time) BETWEEN #{startTime} and #{endTime}

group by referer  order by pv_count desc limit #{size}

    </select>

5.设备终端访问分布接口

 @Override

    public Map<String, List<VisitStatsVO>> queryDeviceInfo(QueryDeviceRequest request) {

        Long accountNo = LoginInterceptor.threadLocal.get().getAccountNo();

        String code = request.getCode();

        String startTime = request.getStartTime();

        String endTime = request.getEndTime();

        String os = QueryDeviceEnum.OS.name().toLowerCase();

        String browser = QueryDeviceEnum.BROWSER.name().toLowerCase();

        String device = QueryDeviceEnum.DEVICE.name().toLowerCase();

        List<VisitStatsDO> osList = visitStatsMapper.queryDeviceInfo(code, accountNo, startTime, endTime, os);

        List<VisitStatsDO> browserList = visitStatsMapper.queryDeviceInfo(code, accountNo, startTime, endTime, browser);

        List<VisitStatsDO> deviceList = visitStatsMapper.queryDeviceInfo(code, accountNo, startTime, endTime, device);

        List<VisitStatsVO> osVisitStatsVOS = osList.stream().map(obj -> beanProcess(obj)).collect(Collectors.toList());

        List<VisitStatsVO> browserVisitStatsVOS = browserList.stream().map(obj -> beanProcess(obj)).collect(Collectors.toList());

        List<VisitStatsVO> deviceVisitStatsVOS = deviceList.stream().map(obj -> beanProcess(obj)).collect(Collectors.toList());

        Map<String, List<VisitStatsVO>> map = new HashMap<>(3);

        map.put("os", osVisitStatsVOS);

        map.put("browser", browserVisitStatsVOS);

        map.put("device", deviceVisitStatsVOS);

        return map;

    }

    <!--查询设备信息分布情况-->

    <select id="queryDeviceInfo" resultMap="BaseResultMap">

        <if test=" field=='os'">

            select os,sum(pv) pv_count from visit_stats where account_no=#{accountNo} and code=#{code} and toYYYYMMDD(start_time) BETWEEN #{startTime} and #{endTime}

            group by os order by pv_count

        </if>

        <if test=" field=='browser'">

            select browser_name,sum(pv) pv_count from visit_stats where account_no=#{accountNo} and code=#{code} and toYYYYMMDD(start_time) BETWEEN #{startTime} and #{endTime}

            group by browser_name order by pv_count

        </if>

        <if test=" field=='device'">

            select device_type,sum(pv) pv_count from visit_stats where account_no=#{accountNo} and code=#{code} and toYYYYMMDD(start_time) BETWEEN #{startTime} and #{endTime}

            group by device_type order by pv_count

        </if>

    </select>

INSERT INTO `default`.visit_stats (code,referer,is_new,account_no,province,city,ip,browser_name,os,device_type,pv,uv,start_time,end_time,ts) VALUES (

'123FNX6a','douyin.com',1,1649837037522,'广东省','广州市','113.68.152.139','Chrome','Windows','COMPUTER',1,1,'2022-05-02 16:17:30.000','2022-05-02 16:17:40.000',1651493856137);

INSERT INTO `default`.visit_stats (code,referer,is_new,account_no,province,city,ip,browser_name,os,device_type,pv,uv,start_time,end_time,ts) VALUES (

'123FNX6a','taobao.com',0,1649837037522,'广东省','广州市','113.68.152.139','Chrome','Windows','COMPUTER',1,0,'2022-05-02 16:17:30.000','2022-05-02 16:17:40.000',1651493856863);

部署上线：

windows本地安装Jenkins

locakhost:8080

密码 667825dc112e4e939d79608c41733032

巴特西

ddddd

项目二阶段总结

账户微服务

短信发送

1.压测发现问题

2.加Async注解实现异步

3.自定义线程池和Async注解配合

对比和问题

4.异步线程池加http连接池

http结构

短信验证码总结

验证码

1.redis连接池

用户注册服务

短链(link)微服务

分库分表

数据库性能优化思路讲解（需要补充详细）

垂直分库

水平分库

垂直分表

水平分表

traffic水平分表

traffic流量表分表总结：

短链业务和实现

重定向

短链码生成

最终方案

短链生成方案总结：

短链信息表的分库分表

短链分组和短链

短链的CRUD

分库分表下的多维度查询问题

冗余双写

短链创建：

异常交换机底层

最终一致性

短链码生成端问题

加分布式锁问题

redis实现分布式锁的问题

流量包商品

防重提交

常见方案

自定义注解防重提交

支付中心

简单工厂加策略模式

https传输过程

第三方支付平台举例：

具体过程

微信支付V3开发前工作

1.配置参数

1.申请APPID：

2.申请mchid：

3.绑定APPID及mchid：

2.配置API key

3.下载并配置商户证书

支付策略接口开发

统一下单接口（unifiedOrder）

开发具体过程

查询订单状态接口（queryPayStatus）

关闭订单接口（closeOrder）

退款接口（refund）

回调验签

验签完的订单状态更新和流量包发放

方案1:

商品微服务用mq：

账号微服务来做流量包权益更新

1.先完成流量包模块的crud

2.配置mq，用来监听商品微服务发送过来的消息，消费者执行业务逻辑

消息队列重复投递

消息队列重复消费

rabbitmq的可靠性投递

rabbit持久化

流量包的过期删除问题和方案

需求：

1.定时任务-xxljob

Redis过期key淘汰策略

2.惰性思想更新流量包

秒杀系统

短链服务和流量包服务