1 课程介绍 1.1 课程背景介绍 每年的春节是中国人的传统节日，大多数中国人都会在这一天选择回家团聚。为了方便用户进行购票， 2012 年春节，铁道部推出 12306网站，进行网络实名购票。然而，历年春节假期，巨大的 访问请求都让中国铁路客户服务中心网站（ www.12306.cn）陷入“万劫不复”。根据新浪的调查，在2013年 春节，有近90%的网友表示12306网站缓慢、页面崩溃，严重影响正常购票。世界级的 人口迁徙带来了一个世界级的难题：要如何通过网络，把火车票及时卖给有需要的人？ 12306网站所面临的问题分析： 铁道部在线车票发售网站12306基本不存在大量图片、视频这些占带宽资源的东西，所面临的主要问题就是 数据库的高并发量——用中国的人口基数来算，这是一个极为恐怖的并发量，在车票 发售的高峰时间点，向12306发起的并发请求数量大得就像一场国家规模的DDOS攻击。 中国铁路客户服务中心网站（www.12306.cn）是世界规模最大的实时交易系统之一，媲美Amazon.com， 节假日尤其是春节的访问高峰，网站压力巨大。据统计，在2012年初的春运高峰期间， 每天有2000万人访问该网站，日点击量最高达到14亿。 而到了2013年春节期间，12306的网站订票系统系统峰值负载达2.6万QPS（每秒钟2.6万次访问请求）或11 万TPS（TPS指每秒服务器处理、传输的事物处理个数，一条消息输入、一条消息输出 、一次数据库访问，均可以折算成TPS），与2012淘宝双11活动峰值负载基本相当。 所以12306所面临的难题本质上也是属于高并发访问问题，类似与一些电商网站所搞的"秒杀"活动一样。通 过对12306的深度优化，2015年12306网站顺利过关，没有“瘫痪”，是值得庆祝的。 而我们本次课题主要就是来探究一下如何对12306网站做深度优化来抵御高并发访问。1.2 高并发访问 那么12306做了什么样的优化，才解决了高并发访问呢 ? 12306技术部主任单杏花在接受一次记者采访的时候有说到：我们研发了分布式的内存计算的余票计算技 术，让余票计算变得非常高效。与此同时单杏花及其团队还研发了 异步交易排队系统， 这种系统采用售取分离、读写分离的核心系统架构等多种技术，为12306售票系统提供技术支撑。 其实通过她的描述，我们可以得出一些处理高并发访问方式： 1、采用内存计算（使用缓存系统） 2、异步处理请求（进行流量消峰） 3、数据库进行读写分离操作 常见的分布式缓存系统：MongoDB ， Redis，MemCache 流量消峰的方案： 1、要对流量进行削峰，最容易想到的解决方案就是用消息队列来缓冲瞬时流量，把同步的直接调用转换成 异步的间接推送，中间通过一个队列在一端承接瞬时的流量洪峰，在另一端平滑地将 消息推送出去。在这里，消息队列就像“水库”一样，拦蓄上游的洪水，削减进入下游河道的洪峰流量，从而 达到减免洪水灾害的目的。 2、答题 你是否还记得，最早期的秒杀只是纯粹地刷新页面和点击购买按钮，它是后来才增加了答题功能的。那么， 为什么要增加答题功能呢？这主要是为了增加购买的复杂度，从而达到两个目的。 第一个目的是防止部分买家使用秒杀器在参加秒杀时作弊。2011年秒杀非常火的时候，秒杀器也比较猖獗， 因而没有达到全民参与和营销的目的，所以系统增加了答题来限制秒杀器。增加 答题后，下单的时间基本控制在2s后，秒杀器的下单比例也大大下降。答题页面如下图所示。 第二个目的其实就是延缓请求，起到对请求流量进行削峰的作用，从而让系统能够更好地支持瞬时的流量高 峰。这个重要的功能就是把峰值的下单请求拉长，从以前的1s之内延长到2s~10s。 这样一来，请求峰值基于时间分片了。这个时间的分片对服务端处理并发非常重要，会大大减轻压力。而 且，由于请求具有先后顺序，靠后的请求到来时自然也就没有库存了，因此根本到不 了最后的下单步骤，所以真正的并发写就非常有限了。 3、分时间段进行产品上架处理 其实处理高并发访问还有两种常见手段：静态化、集群 静态化：分布式缓存是为了解决数据库服务器和Web服务器之间的瓶颈，如果一个网站流量很大这个瓶颈将会非常明 显，每次数据库查询耗费的时间将不容乐观。对于更新速度不是很快的站点，可以 采用静态化来避免过多的数据查询，可使用Freemaker或Velocity来实现页面静态化。 集群: 使用多台服务器去处理并发请求 1.3 系统架构介绍 基于以上几点高并发的处理方案，我们本次所设计的12306后端系统架构如下所示： 1.3.1 数据同步架构 系统管理员通过后台管理系统基于一些基础数据（座位数据，列车车次数据，乘车计划数据）生成指定日期 的乘车计划数据。然后我们通过logstash将生成的数据同步到ES和Redis中。 Logstash常见的数据获取方式拉，推。上述架构给大家展示的是拉的模式，但是这种方式我们当前这个系统 环境中不太适合，原因是因为我们使用了MyCat进行分库分表的处理，而 Logstash在进行拉取数据的时候如果数据量较大我们就需要进行分页拉取，那么此时Logstash就会生成类似 这样的一条sql语句：select count(*) as count from ....来查询满足条件 总条数，但是这个count别名使用了反引号，而这个反引号在MyCat中无法使用，因此就会产生异常。 因此本次我们在进行数据同步的时候使用的是Logstash的推模式进行数据同步，如下所示： 1.3.2 数据搜索架构 数据同步完毕以后，用户就可以搜索相关的乘车计划数据了。具体的搜索架构如下所示： 1.3.3 用户下单架构 通常订票系统要处理生成订单、减扣库存、用户支付这三个基本的阶段，我们系统要做的事情是要保证火车 票订单不超卖、不少卖，每张售卖的车票都必须支付才有效，还要保证系统承受极高 的并发。 这三个阶段的先后顺序改怎么分配才更加合理呢？ 第一种方案: 当用户并发请求到达服务端时，首先创建订单，然后扣除库存，等待用户支付。这种顺序是我们一般人首先 会想到的解决方案，这种情况下也能保证订单不会超卖，因为创建订单之后就会减 库存，这是一个原子操作。 存在的问题： 第一就是在极限并发情况下，任何一个内存操作的细节都至关影响性能，尤其像创建订单这种逻辑，一般都 需要存储到磁盘数据库的，对数据库的压力是可想而知的； 第二是如果用户存在恶意下单的情况，只下单不支付这样库存就会变少，会少卖很多订单。 第二种方案： 如果等待用户支付了订单在减库存，第一感觉就是不会少卖。但是这是并发架构的大忌，因为在极限并发情 况下，用户可能会创建很多订单，当库存减为零的时候很多用户发现抢到的订单支付 不了了，这也就是所谓的"超卖", 并且这种方案也不能避免并发操作数据库磁盘IO。 第三种方案： 从上边两种方案的考虑，我们可以得出结论：只要创建订单，就要频繁操作数据库IO。那么有没有一种不需 要直接操作数据库IO的方案呢，这就是预扣库存。先扣除了库存，保证不超卖，然后 异步生成用户订单，这样响应给用户的速度就会快很多；那么怎么保证不少卖呢？用户拿到了订单，不支付 怎么办？我们都知道现在订单都有有效期，比如说用户五分钟内不支付，订单就失效 了，订单一旦失效，就会加入新的库存，这也是现在很多网上零售企业保证商品不少卖采用的方案。订单的 生成是异步的,一般都会放到MQ（消费队列）中处理,订单量比较少的情况下，生成订 单非常快，用户几乎不用排队。如下图所示： 这种方案也就是单杏花主任所提出的异步交易排队系统。当然12306网站的还有一个改造的关键技术 建立可 伸缩扩展的云应用平台。根据互联网上的新闻，中国铁道科学研究院电子计算技术研究所副所长，12306网站技术负责人朱建生说， 为了应对2015年春运售票高峰，该网站采取5项措施： 一、利用外部云计算资源分担系统查询业务，可根据高峰期业务量的增长按需及时扩充。 二、对系统的互联网接入带宽进行扩容，并可根据流量情况快速调整，保证高峰时段旅客顺畅访问网站。 三、防范恶意抢票，通过技术手段屏蔽抢票软件产生的恶意流量，保证网站健康运行，维护互联网售票秩 序。 四、制定了多套应急预案，以应对突发情况。 2 用户下单 2.1 下单流程 用户下单排队的具体流程如下所示： 2.2 下单页面 注意：下单之前需要初始化一些用户数据到Redis中(train_manager:userInfo)，默认用户已经登录 下单页面：12306\otn\confifirmPassenger\initDc.html 2.3 订单服务实现 2.3.1 下单接口定义 实体类创建 OrderParams（接收下单请求参数的实体类） Order（订单实体类） @Data // 通过lombok生成getter和 setter方法 @NoArgsConstructor // 通过lombok生成无参构造方法 @ToString // 通过lombok生成toString方法 @EqualsAndHashCode // 通过lombok生成hashCode和 equals方法 public class OrderParams { private String ticketType ; // 座位类型 private Long ridingPlanId ; private String trainNum ; private String trainRidingDate ; private String userId ; } 1 2 3 4 5 6 7 8 9 10 11 12 13 @Data // 通过lombok生成getter和 setter方法 @NoArgsConstructor // 通过lombok生成无参构造方法 @ToString // 通过lombok生成toString方法 @EqualsAndHashCode // 通过lombok生成hashCode和 equals方法 public class Order { private Long id ; private Long ridingPlanDateId ; private String startStationName ; private String receiveStationName ; private String trainNum ; private String trainRidingDate ; private String startTime ; private String endTime ; private String seatNum ; private String seatNature ; // 座位性质 private String coachNum ; // 车厢号 private Double seatPrice ; 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 ResponseResult（用于封装下单结果） 订单工程环境搭建 1、导入下单工程（itheima-train-manager-order） 2、在nacos配置中心中添加配置项（itheima-train-manager-order-dev.yml） 下单接口定义 private String payStatus ; // 支付状态，0表示未支付，1表示 支付 private String userId ; private String isEffect ; // 是否有效：1表示有效，0表示无效 } 19 20 21 22 23 /** * 响应结果 */ @Data // 通过lombok生成getter和setter方法 @NoArgsConstructor // 通过lombok生成无参构造方法 @ToString // 通过lombok生成toString方法 @EqualsAndHashCode // 通过lombok生成hashCode和equals方法 public class ResponseResult<T> { private boolean result ; private String message ; private T data ; } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 spring: application: name: itheima-train-manager-order redis: cluster: nodes: 172.17.0.224:6379,172.17.0.224:6380,172.17.0.224:6381,172.17.0.224:6382, 172.17.0.224:6383,172.17.0.224:6384 server: port: 9056 1 2 3 4 5 6 7 8 @RestController @RequestMapping(value = "/order") public class OrderController { @Autowired private OrderService orderService ; @RequestMapping (value = "/submitOrder" ) public ResponseResult < String > submitOrder (@RequestBody OrderParams orderParams ) { return orderService.submitOrder(orderParams) ; } } 1 2 3 4 5 6 7 8 9 10 11 12 13 2.3.2 Nginx配置 反向代理配置 在hosts文件中配置www.trainmanager.order.com域名映射 # 下单服务的代理转发地址 upstream train-manager-order { server 127.0.0.1:9056 ; } # 配置下单工程的反向代理 server { listen 80; server_name www.trainmanager.order.com; access_log logs/host.access.order.log main ; # nginx访问日志 location / { proxy_pass http://train-manager-order ; } error_page 500 502 503 504 /50x.html; location = /50x.html { root html; } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 Nginx限流配置 为了防止一些抢票助手所发起的一些无用请求，我们可以使用nginx中的限流策略进行限流操作。 常见的限流算法 常见的限流算法有一下的几种：计数器、漏桶算法、令牌桶算法 计数器 计数器法是限流算法里最简单也是最容易实现的一种算法。比如我们规定，对于A接口来说，我们1分钟的访 问次数不能超过100个。那么我们我们可以设置一个计数器counter， 其有效时间为1分钟（即每分钟计数器会被重置为0），每当一个请求过来的时候，counter就加1，如果 counter的值大于100，就说明请求数过多；如下图所示： 这个算法虽然简单，但是有一个十分致命的问题，那就是临界问题。 如下图所示，在1:00前一刻到达100个请求，1:00计数器被重置，1:00后一刻又到达100个请求，显然计数器 不会超过100，所有请求都不会被拦截；然而这一时间段内请求数已 经达到200，远超100。 漏桶算法 漏桶算法其实很简单，可以粗略的认为就是注水漏水过程，往桶中以一定速率流出水，以任意速率流入水， 当水超过桶流量则丢弃，因为桶容量是不变的，保证了整体的速率。 如下图所示： 令牌桶算法 令牌桶是一个存放固定容量令牌的桶，按照固定速率r往桶里添加令牌；桶中最多存放b个令牌，当桶满时， 新添加的令牌被丢弃；当一个请求达到时，会尝试从桶中获取令牌； 如果有，则继续处理请求；如果没有则排队等待或者直接丢弃；可以发现，漏桶算法的流出速率恒定，而令 牌桶算法的流出速率却有可能大于r； 从作用上来说，漏桶和令牌桶算法最明显的区别就是是否允许突发流量(burst)的处理，漏桶算法能够强行限 制数据的实时传输（处理）速率，对突发流量不做额外处理；而令牌桶算法能够在限制数据的平均传输速率的同时允许某种程度的突发传输。 Nginx的限流策略 Nginx的限流主要是两种方式：限制访问频率和限制并发连接数。 Nginx按请求速率限速模块使用的是漏桶算法，即能够强行保证请求的实时处理速度不会超过设置的阈值。 Nginx官方版本限制IP的连接和并发分别有两个模块： 1、limit_req_zone：用来限制单位时间内的请求数，即速率限制 , 采用的漏桶算法 "leaky bucket"。 2、limit_conn_zone：用来限制同一时间连接数，即并发限制。 limit_req_zone 使用语法：limit_req_zone key zone rate key ： 定义限流对象，binary_remote_addr 是一种key，表示基于 remote_addr(客户端IP) 来做限流， binary_ 的目的是压缩内存占用量。 zone：定义共享内存区来存储访问信息， myRateLimit:10m 表示一个大小为10M，名字为 myRateLimit的内存区域。1M能存储16000 IP地址的访问信息，10M可以存储16W IP地址 访问信息。 rate 用于设置最大访问速率，rate=10r/s 表示每秒最多处理10个请求。Nginx 实际上以毫秒为粒度来跟 踪请求信息，因此 10r/s 实际上是限制：每100毫秒处理一个请求。这意味着， 自上一个请求处理完后，若后续100毫秒内又有请求到达，将拒绝处理该请求。 举例： limit_conn_zone 使用语法：limit_conn_zone key zone key ： 定义限流对象，binary_remote_addr 是一种key，表示基于 remote_addr(客户端IP) 来做限流， binary_ 的目的是压缩内存占用量。 zone：定义共享内存区来存储访问信息， myRateLimit:10m 表示一个大小为10M，名字为 myRateLimit的内存区域。1M能存储16000 IP地址的访问信息，10M可以存储16W IP地址 访问信息。 举例： http { # 定义限流策略 limit_req_zone $binary_remote_addr zone=rateLimit:10m rate=1r/s ; # 搜索服务的虚拟主机 server { location / { # 使用限流策略，burst=5，重点说明一下这个配置，burst爆发的意思，这个 配置的意思是设置一个大小为5的缓冲区（队列）当有大量请求（爆发）过来时， # 超过了访问频次限制的请求可以先放到这个缓冲区内。nodelay，如果设置， 超过访问频次而且缓冲区也满了的时候就会直接返回503，如果没有设置，则所 # 有请求会等待排队。 limit_req zone=rateLimit burst=5 nodelay; proxy_pass http://train-manager-search ; } } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 http { # 定义限流策略 limit_conn_zone $binary_remote_addr zone=perip:10m; 1 2 3 4 limit_conn_zone $server_name zone=perserver:10m; # 搜索服务的虚拟主机 server { location / { # 对应的key是 $binary_remote_addr，表示限制单个IP同时最多能持有1 个连接。 limit_conn perip 1; # 对应的key是 $server_name，表示虚拟主机(server) 同时能处理并发连 接的总数。注意，只有当 request header 被后端server处理后，这个连接才进行计数。 limit_conn perserver 10 ; proxy_pass http://train-manager-search ; } } } 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 JMeter测试工具 为了模拟并发访问，我们可以使用一下JMeter这个压测工具。 JMeter工具下载 我们可以登录JMeter 的官网来下载JMeter： http://jmeter.apache.org/download_jmeter.cgi JMeter使用 1、启动JMeter：执行bin目录下的jmeter.bat 2、在测试计划中添加一个线程组 3、在指定的线程组中创建一个Http请求，其中线程的数量表示并发的数据量 4、如果需要发送post请求，并且传递json数据。那么我们还需要该给http请求添加一个配置单元(http信息 头管理器)，指定请求数据的格式是json Content-Type: application/json 测试数据: 5、添加请求监听器：查看结果树 6、执行测试，查看Nginx的日志记录 测试方案 方案一：测试limit_req_zone,注释掉其他的两个限流方式 方案二：测试limit_conn_zone，注释请求次数限流策略 { "ridingPlanId":264546196624769024, "ticketType":"secondSeat", "trainNum":"G26", "trainRidingDate":"2020-06-01", "userId":"fb71889cbfbf4857a32f701f175fe256" } 1 2 3 4 5 6 7 2.3.3 生成订单 预扣库存 // 依赖注入 @Autowired private StringRedisTemplate redisTemplate ; // 下单操作 public ResponseResult<String> submitOrder(OrderParams orderParams) { // 定义ResponseResult返回对象 ResponseResult<String> responseResult = new ResponseResult<>() ; // 获取相关参数 String trainNum = orderParams.getTrainNum() ; String trainRidingDate = orderParams.getTrainRidingDate(); Long ridingPlanId = orderParams.getRidingPlanId(); String ticketType = orderParams.getTicketType(); // 座位的性质 String redisticketTypeKey = ticketType + "Stock" ; // busSeat busSeatStock 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 // 构建乘车计划的key // riding_plan:G26:2020-06-01:270101019289976832 LOGGER.info("【预扣库存开始了】 ---> " + JSON.toJSONString(orderParams)); String ridingPlanDateKey = "riding_plan:" + trainNum + ":" + trainRidingDate + ":" + ridingPlanId ; int count = Integer.parseInt (redisTemplate.boundHashOps(ridingPlanDateKey).get(redi sticketTypeKey ). toString ()) ; if(count <= 0 ) { responseResult.setResult(false); return responseResult ; } // 动态库存扣减 String stockCountKeyDynamic = "riding_plan:" + trainNum + ":" + trainRidingDate + ":*" ; Set<String> keys = redisTemplate.keys(stockCountKeyDynamic); for(String key : keys) { redisTemplate.boundHashOps(key).put(redisticketTypeKey , String.valueOf(count - 1)); } LOGGER.info("【预扣库存结束了】 ---> " + JSON.toJSONString(orderParams)); return null ; } 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 分配座位 具体的实现步骤如下所示： 1、获取所有的指定座位类型的所有车厢的key（集合） 2、遍历集合获取每一个车厢数据（集合） 3、遍历集合获取每一个座位所对应的状态值 4、状态值如果是0，记录座位标号 5、更改座位状态值 具体的代码实现如下所示： // 定义前端参数和座位性质之间的Map集合 private static Map<String , String> seatNatureHashMap = new HashMap<String , String>() ; static { seatNatureHashMap.put("busSeat" , "2") ; seatNatureHashMap.put("firstSeat" , "3") ; seatNatureHashMap.put("secondSeat" , "4") ; seatNatureHashMap.put("hardSleeper" , "7") ; seatNatureHashMap.put("softSleeper" , "6") ; seatNatureHashMap.put("hardSeat" , "5") ; seatNatureHashMap.put("softSeat" , "9") ; } // 依赖注入 @Autowired private StringRedisTemplate redisTemplate ; 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 // 下单操作 public ResponseResult<String> submitOrder(OrderParams orderParams) { ... // 获取前端所传递的参数或者后端座位类型 LOGGER.info ("【分配座位开始了】 ---> " + JSON.toJSONString (orderParams )); String seatNature = seatNatureHashMap .get(ticketType); // 分配座位 : train_seat:C1002:2020-06-01:4:6 // 1. 获取所有的指定座位类型的所有车厢的key String seatNatureKey = "train_seat:" + trainNum + ":" + trainRidingDate + ":" + seatNature + ":*" ; Set<String> allSeatNatureCoach = redisTemplate.keys(seatNatureKey); // 2. 遍历集合 String coach = null ; String seatNo = null; out: for(String key : allSeatNatureCoach) { // 获取每一个车厢数据 Map<Object, Object> entries = redisTemplate.boundHashOps(key).entries(); coach = entries.get("coach_num").toString() ; // 遍历集合 for(Object seatNoKey : entries.keySet()) { // 获取每一个座位所对应的状态值 String seatStatus = entries.get(seatNoKey).toString(); // 状态值如果是0，记录座位标号 if("0".equals(seatStatus)) { seatNo = seatNoKey.toString() ; redisTemplate.boundHashOps(key).put(seatNoKey.toString() , "1"); break out; } } } 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 生成订单 为了提高数据响应速度，我们在构建订单数据的时候可以使用一个线程来完成。并且在该线程执行完毕以 后，要获取执行结果，因此这个任务类我们需要使用Callable，执行这个任务的 线程我们可以使用线程池来完成。具体的实现步骤如下所示： 1、创建雪花算法对应的类对象 2、创建线程池对象 3、获取指定日期的乘车计划数据，并将其封装到一个TbRidingPlanDate对象中 4、生成订单 具体的代码实现如下所示： LOGGER.info("【分配座位结束了】 ---> coach ---> {} , seatNo ---> {} " , coach , seatNo); return null ; } 56 57 58 59 // 创建雪花算法对应类对象 private SnowFlakeGenerator snowFlakeGenerator = new SnowFlakeGenerator( 0, 2) ; // 创建一个线程池 private ExecutorService executorService = Executors.newFixedThreadPool(10) ; // 依赖注入 1 2 3 4 5 6 7 @Autowired private StringRedisTemplate redisTemplate ; // 下单操作 public ResponseResult<String> submitOrder(OrderParams orderParams) { ... // 获取 ridingPlanDateKey所对应的日期乘车计划数据 Map< Object , Object> ridingPlanDateHashMap = redisTemplate .boundHashOps (ridingPlanDateKey). entries (); TbRidingPlanDate tbRidingPlanDate = JSON.parseObject(JSON.toJSONString(ridingPlanDateHashMap), TbRidingPlanDate.class ); ... // 生成订单数据 String finalCoachNum = coach ; String finalSeatNum = seatNo ; Callable<ResponseResult<String>> callable = () -> { // 创建订单对象 LOGGER.info("【构建订单开始了】 ---> " + JSON.toJSONString(orderParams)); Long orderId = snowFlakeGenerator.nextId() ; // 构建订单数据 Order order = new Order() ; order.setId(orderId); order.setCoachNum(finalCoachNum); order.setIsEffect("1"); order.setPayStatus("0"); order.setRidingPlanDateId(orderParams.getRidingPlanId()); order.setSeatNature(seatNature); order.setSeatNum(finalSeatNum); order.setTrainNum(orderParams.getTrainNum()); order.setUserId(orderParams.getUserId()); order.setTrainRidingDate(orderParams.getTrainRidingDate()); order.setStartStationName(tbRidingPlanDate.getStartStationName()); order.setReceiveStationName(tbRidingPlanDate.getReceiveStationName()); 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 if ("终点站".equals(tbRidingPlanDate.getStartTime())) { order.setStartTime(tbRidingPlanDate.getReceiveStartStationTime()); } else { order.setStartTime(tbRidingPlanDate.getStartTime()); } order .setEndTime(tbRidingPlanDate.getReceiveEndStationTime()); // 根据座位情况设置具体的票价 if ( "7".equals(seatNature)) { // 硬卧 if (finalSeatNum.contains("上铺")) { order.setSeatPrice(tbRidingPlanDate.getHardSleeperUpPrice()); } else if (finalSeatNum.contains("中铺")) { order.setSeatPrice(tbRidingPlanDate.getHardSleeperMiddlePrice()); } else if (finalSeatNum.contains("下铺")) { order.setSeatPrice(tbRidingPlanDate.getHardSleeperDownPrice()); } } else if ("6".equals(seatNature)) { // 软卧 if (finalSeatNum.contains("上铺")) { order.setSeatPrice(tbRidingPlanDate.getSoftSleeperUpPrice()); } else if (finalSeatNum.contains("下铺")) { order.setSeatPrice(tbRidingPlanDate.getSoftSleeperDownPrice()); } } else if ("3".equals(seatNature)) { order.setSeatPrice(tbRidingPlanDate.getFirstSeatPrice()); } else if ("4".equals(seatNature)) { order.setSeatPrice(tbRidingPlanDate.getSecondSeatPrice()); } else if ("2".equals(seatNature)) { order.setSeatPrice(tbRidingPlanDate.getBusSeatPrice()); } else if ("5".equals(seatNature)) { order.setSeatPrice(tbRidingPlanDate.getHardSeatPrice()); } else if ("9".equals(seatNature)) { order.setSeatPrice(tbRidingPlanDate.getSoftSeatPrice()); } LOGGER.info("【构建订单结束了】 ---> " + JSON.toJSONString(order)); responseResult.setResult(true); 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 Redis排队 return responseResult ; } ; try { Future <ResponseResult <String >> future = executorService .submit(callable ); ResponseResult <String> stringResponseResult = future.get(); return stringResponseResult ; } catch (Exception e) { e.printStackTrace(); } return null ; } 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 采用Redis中的ZSet集合存储排队信息，使用：列车编号 + 乘车日期 + 用户id作为key，使用当前系统时间的 纳秒值作为value LOGGER.info("【Redis排队开始了】 ---> " + JSON.toJSONString(order)); String sortedKey = trainNum + ":" + trainRidingDate + ":" + orderParams.getUserId () ; redisTemplate .boundZSetOps ("train_manager:sorted_queue").add(sortedKey , System.nanoTime ()); LOGGER.info ("【Redis 排队结束了】 ---> " + JSON.toJSONString(order)); 1 2 3 4 2.3.4 同步ES库存 发送同步数据 那么我们首先来完成一下生产者的代码，具体的步骤如下所示： 1、定义要同步的数据的JavaBean 2、添加消息队列的相关配置 nacos配置中心添加rabbitmq的配置信息 @Data // 通过lombok生成getter和setter方法 @NoArgsConstructor // 通过lombok生成无参构造方法 @ToString // 通过lombok生成toString方法 @EqualsAndHashCode // 通过lombok生成hashCode和equals方法 public class OrderSynEs { private String ridingPlanDate ; // 乘车日期 private String seatNature ; // 座位性质 private String trainNum ; // 列车车次 } 1 2 3 4 5 6 7 8 9 10 11 RabbitmqConfifigur ation配置类添加如下配置： 3、定义发送数据的service（直接复制itheima-train-manager工程中的RabbitmqProducer） 4、更改OrderService构建OrderSynEs对象，发送同步数据 rabbitmq: host: 172.17.0.224 port: 5672 username: admin password: admin virtual-host : / publisher-confirm-type : correlated # 设置生产者通道支持confirm模式 1 2 3 4 5 6 7 // 声明队列 @Bean(name = "syn_stock_es_queue") public Queue synOrderStockEsQueue() { return QueueBuilder.durable("syn_stock_es_queue").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBindOrderStockEsQueue(@Qualifier(value = "train_manager_ex") Exchange exchange , @Qualifier(value = "syn_stock_es_queue") Queue queue) { return BindingBuilder.bind(queue).to(exchange).with("syn_stock").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 LOGGER.info("【发送ES同步库存数据开始了】 ---> " + JSON.toJSONString(order)); OrderSynEs orderSynEs = new OrderSynEs() ; orderSynEs.setRidingPlanDate(trainRidingDate); orderSynEs.setSeatNature(seatNature); orderSynEs.setTrainNum(trainNum); rabbitmqProducer.sendMessage(JSON.toJSONString(orderSynEs) , "syn_stock"); LOGGER.info("【发送ES同步库存数据结束了】 ---> " + JSON.toJSONString(orderSynEs)); 1 2 3 4 5 6 7 接收同步数据 接下来我们就需要在itheima-train-manager-es同步数据工程中来接收同步数据，更新ES的库存信息。具体 的实现步骤如下所示： 1、在RabbitmqConfifiguration配置类添加如下配置： 2、在EsConsumer类中添加监听同步库存队列的方法，方法代码逻辑如下所示： // 声明队列 @Bean(name = "syn_stock_es_queue") public Queue synOrderStockEsQueue() { return QueueBuilder.durable("syn_stock_es_queue").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBindOrderStockEsQueue(@Qualifier(value = "train_manager_ex" ) Exchange exchange , @Qualifier(value = "syn_stock_es_queue" ) Queue queue ) { return BindingBuilder.bind(queue).to(exchange).with("syn_stock").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 @RabbitListener(queues = "syn_stock_es_queue") public void synStock(String orderSynEsJson , Message message , Channel channel) { try { // 解析JSON数据 OrderSynEs orderSynEs = JSON.parseObject(orderSynEsJson, OrderSynEs.class); esIndexService.synStockInfo(orderSynEs) ; logger.info("【同步下单扣减库存数据成功】---->" + orderSynEsJson); // 自动应答 channel.basicAck(message.getMessageProperties().getDeliveryTag() , false); }catch (Exception e) { e.printStackTrace(); logger.error("【同步下单扣减库存数据失败】---->" + orderSynEsJson); } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 ESIndexService类添加如下方法： public void synStockInfo(OrderSynEs orderSynEs) throws IOException { // 创建搜索请求对象 SearchRequest searchRequest = new SearchRequest("riding_plan_date") ; SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder ().trackTotalHits (true) ; // 表示要查询所有的数据 BoolQueryBuilder boolQueryBuilder = QueryBuilders .boolQuery (); boolQueryBuilder .must (QueryBuilders . termQuery("trainNum" , orderSynEs.getTrainNum())) ; boolQueryBuilder.must(QueryBuilders.termQuery("trainRidingDate" , orderSynEs.getRidingPlanDate())) ; searchSourceBuilder.query(boolQueryBuilder); searchRequest.source(searchSourceBuilder) ; // 进行搜索 SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); SearchHits searchHits = searchResponse.getHits(); SearchHit[] hitsHits = searchHits.getHits(); for (SearchHit hitsHit : hitsHits) { String esSearchResult = hitsHit.getSourceAsString(); TbRidingPlanDate ridingPlanDate = JSON.parseObject(esSearchResult, TbRidingPlanDate.class); switch (orderSynEs.getSeatNature()) { case "3" : ridingPlanDate.setFirstSeatStock(ridingPlanDate.getFirstSeatStock() - 1); break ; case "4": ridingPlanDate.setSecondSeatStock(ridingPlanDate.getSecondSeatStock() - 1); break; case "7": ridingPlanDate.setHardSleeperStock(ridingPlanDate.getHardSleeperStock( ) - 1); break ; case "6": 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 ridingPlanDate.setSoftSleeperStock(ridingPlanDate.getSoftSleeperStock( ) - 1); break; case "2": ridingPlanDate .setBusSeatStock(ridingPlanDate.getBusSeatStock() - 1); break ; case "5": ridingPlanDate .setHardSeatStock(ridingPlanDate.getHardSeatStock() - 1); break ; case "9": ridingPlanDate.setSoftSeatStock( ridingPlanDate.getSoftSeatStock() - 1); break ; } // 创建更新请求对象 UpdateRequest updateRequest = new UpdateRequest("riding_plan_date" , hitsHit.getId()).doc(JSON.toJSONString(ridingPlanDate) , XContentType.JSON) ; restHighLevelClient.update(updateRequest , RequestOptions.DEFAULT) ; } } 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 2.3.5 发送订单数据 实现思路 按照我们最初的想法，我们只需要将订单Order对象发送到Mq中，然后订单处理服务从队列中监听到Order 数据生成订单即可。但是我们后续还有另外一个操作，就是在下单成功以后，我们需要将 下单状态通过websocket推送给用户，因此我们需要在订单处理服务中不单单需要获取到订单数据，还需要 获取到用户的数据。因此在发送下单数据的时候单单发送订单的数据时不够的。我们需要 创建一个实体类，用来封装订单以及该订单所对应的用户数据（OrderHandler）。 代码实现 具体的实现步骤如下所示： 1、 创建实体类。OrderHandler（用于封装发送给MQ的订单数据） 2、 RabbitmqConfifiguration配置类添加如下配置： @Data // 通过lombok生成getter和setter方法 @NoArgsConstructor // 通过lombok生成无参构造方法 @ToString // 通过lombok生成toString方法 @EqualsAndHashCode // 通过lombok生成hashCode和equals方法 public class OrderHandler { private Order order ; // 订单信息 private User user ; // 用户信息 } 1 2 3 4 5 6 7 8 9 10 3、更改OrderService发送订单数据 // 声明队列 @Bean(name = "syn_gen_order_queue") public Queue genOrderQueue() { return QueueBuilder.durable("syn_gen_order_queue").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBindGenOrderQueue(@Qualifier(value = "train_manager_ex" ) Exchange exchange , @Qualifier(value = "syn_gen_order_queue" ) Queue queue ) { return BindingBuilder.bind(queue).to(exchange).with("gen_order").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 // 发送订单数据到MQ中 LOGGER.info("【发送订单数据开始了】 ---> " + JSON.toJSONString(order)); User user = JSON.parseObject(redisTemplate.boundHashOps("train_manager:userInfo").ge t(orderParams.getUserId()).toString() , User.class) ; OrderHandler orderHandler = new OrderHandler() ; orderHandler.setUser(user); orderHandler.setOrder(order); rabbitmqProducer.sendMessage(JSON.toJSONString(orderHandler) , "gen_order"); LOGGER.info("【发送订单数据开始了】 ---> " + JSON.toJSONString(orderHandler)); 1 2 3 4 5 6 7 8 2.3.6 查询排队接口 当用户下单完毕以后，就会跳转到下单成功页面。那么在下单成功页面就需要调用查询排队接口，去查询排 队信息。 接口定义： 业务逻辑实现 @RequestMapping (value = "/queryQueue") public ResponseResult<String> querySortedInfo(@RequestBody OrderParams orderParams) { return orderService.querySortedInfo(orderParams) ; } 1 2 3 4 // 查询排队信息 public ResponseResult<String> querySortedInfo(OrderParams orderParams) { // 创建ResponseResult对象 ResponseResult<String> responseResult = new ResponseResult<>() ; // 获取排队信息 String sortedKey = orderParams.getTrainNum() + ":" + orderParams.getTrainRidingDate() + ":" + orderParams.getUserId() ; Long rank = redisTemplate.boundZSetOps("train_manager:sorted_queue").rank(sortedKey ); // 判断结果 if(rank == null) { responseResult.setResult(false); }else { responseResult.setResult(true); responseResult.setData(String.valueOf(rank)); } // 返回 return responseResult ; 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 } 23 2.4 订单处理服务 2.4.1 获取订单队列数据 我们首先来完成以下订单处理服务获取订单数据的代码。具体的实现步骤如下所示： 1、导入订单处理服务工程 (itheima-train-manager-order-handler) 2、在nacos配置中心中创建订单处理服务的配置（ itheima-train-manager-order-handler-dev.yml） 3、添加消息监听代码 spring: application: name: itheima-train-manager-order-handler rabbitmq: host: 172.17.0.224 port: 5672 username: admin password: admin virtual-host: / listener: simple: acknowledge-mode: manual # 设置消息手动应答 # 配置数据源 datasource: driver-class-name: com.mysql.jdbc.Driver url: jdbc:mysql://172.17.0.224:3366/db_train username: root password: 1234 # 配置mybatis的信息 mybatis: type-aliases-package: com.itheima.train.manager.domain mapper-locations: classpath:com/itheima/train/manager/order/handler/mapper/*Mapper.xml server: port: 10001 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 @Component public class OrderHandlerConsumer { 1 2 3 // 定义日志记录器 private Logger LOGGER = LoggerFactory.getLogger(OrderHandlerConsumer.class) ; @RabbitListener(queues = "syn_gen_order_queue") public void orderHanderConsumer (String orderHandlerJsonInfo , Message message , Channel channel) { try { // 解析数据 OrderHandler orderHandler = JSON.parseObject(orderHandlerJsonInfo, OrderHandler.class); // 调用service进行数据保存 LOGGER.info("【保存订单数据完成】 ---> " + orderHandlerJsonInfo); // 给出应答 channel.basicAck(message.getMessageProperties().getDeliveryTag() , false); } catch (IOException e) { e.printStackTrace(); LOGGER.info("【保存订单数据失败】 ---> " + orderHandlerJsonInfo); } } } 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 2.4.2 订单数据库架构 订单数据特点：写并发量大于读并发量 如何提高我们写数据的能力，给用户良好的用户体验，就是我们需要研究的目标！ 设计方向： 1、多个节点进行数据写入 2、进行读写分离操作，提高单节点写数据的并发能力 3、要保证每一个写入节点的高可用，当主节点出现问题以后，从节点立马升级为主节点 基于以上几点的设计思路，我们所设计出来的订单数据库的架构如下所示： 2.4.3 MySql主从复制 主从复制简介 就是有两个数据库服务器，一个是主（ master）数据库服务器，另一个是从（slave）数据库服务器。当主 （master）数据库有数据写入，包括插入、删除、 修改，都会在从（slave）数据库上操作一次。这样的操作下，主从（slave）数据库的数据都是一样的，就 相当于时刻在做数据备份，就算主（master）数据 库的数据全部丢失了，还有从（slave）数据库的数据，我们就可以把从（slave）数据库的数据导出来进行 数据恢复。 主从复制原理 主从复制原理主要有三个线程不断在工作： 1、主（master）数据库启动bin二进制日志，这样会有一个Dump线程，这个线程是把主（master）数据 库的写入操作都会记录到这个bin的二进制文件中。 2、然后从（slave）数据库会启动一个I/O线程，这个线程主要是把主（master）数据库的bin二进制文件读 取到本地，并写入到中继日志（Relay log）文件中。 3、最后从（slave）数据库其他SQL线程，把中继日志（Relay log）文件中的事件再执行一遍，更新从 （slave）数据库的数据，保持主从数据一致。 2.4.4 一主一从介绍 一主一从部署 在Mycat中，读写分离可以说有两种：一种是一主一从，另一种是多主多从。接下来我们就来首先给大家介 绍一下一主一从的部署。 分为两步进行实现： 1、一主一从的数据库配置实现 2、MyCat一主一从实现读写分离配置 一主一从的MySql部署步骤如下所示： 1、在指定的目录下创建对应的文件夹（做mysql容器的目录映射） 2、将docker中的mysql配置文件分别复制到mysql10/confifig和mysql11/confifig目录中 3、对两个节点的配置文件（/usr/local/mysql-master-slave/mysql10/confifig/mysql.conf.d）做如下配置： 4、创建两个mysql的容器 mysql-master-slave mysql10 config data mysql11 config data 1 2 3 4 5 6 7 lower_case_table_names=1 # 不区分字母大小写 log-bin=mysql-bin # 开启二进制日志 server-id=10 # 设置server-id , 从节点指定一个其他的id即可 # 不同步哪些数据库 binlog-ignore-db = mysql binlog-ignore-db = test binlog-ignore-db = information_schema 1 2 3 4 5 6 7 8 9 5、进入到mysql10容器中，主节点状态 其中File和Position都是我们在设置从（slave）数据库的时候用到的； 6、登录到从数据库，进行以下配置。 配置主（master）数据库的IP地址，用户名，登录密码，刚才在主（master）数据库中查到的bin二进制文 件的名称和所在的位置。 docker run -di --network=docker-network --ip=172.19.0.110 -- name=train_mysql_01 -p 3410:3306 --privileged=true -e MYSQL_ROOT_PASSWORD=1234 -v /usr/local/mysql-master slave/mysql10/data:/var/lib/mysql -v /usr/local/mysql-master slave/mysql10/config:/etc/mysql/ mysql:5.6 docker run -di --network =docker-network --ip= 172 .19.0.111 -- name=train_mysql_02 -p 3411 :3306 --privileged = true -e MYSQL_ROOT_PASSWORD =1234 -v /usr/local/mysql-master slave/mysql11/data:/var/lib/mysql -v /usr/local/mysql-master slave/mysql11/config:/etc/mysql/ mysql:5.6 1 2 mysql> show master status ; +------------------+----------+--------------+-------------------------- -----+-------------------+ | File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set | +------------------+----------+--------------+-------------------------- -----+-------------------+ | mysql-bin.000004 | 120 | | mysql,test,information_schema | | +------------------+----------+--------------+-------------------------- -----+-------------------+ 1 row in set (0.00 sec) mysql> 1 2 3 4 5 6 7 8 9 mysql> stop slave; # 先关闭从节点 Query OK, 0 rows affected, 1 warning (0.00 sec) mysql> change master to master_host='172.19.0.110', master_user='root', master_password='1234', master_log_file='mysql-bin.000004', master_log_pos=120; Query OK, 0 rows affected, 2 warnings (0.34 sec) mysql> start slave; Query OK, 0 rows affected (0.00 sec) mysql> show slave status\G # 查看从节点状态 1 2 3 4 5 6 7 一主一从测试 接下来我们就来对刚才所部署的一主一从架构的数据库进行测试。具体的测试步骤如下所示： 1、使用Navicat连接mysql10和mysql11 2、在mysql10节点上创建一个数据库db_train，然后查看从数据库的变化 3、在主节点上db_train数据库中创建表以及插入数据，然后查看从数据库的变化 MyCat读写分离 接下来我们就来研究一下基于一主一从架构的mysql，要实现读写分离我们的mycat应该如何进行配置。 1、配置server.xml文件（给指定的用户添加逻辑库） 2、mycat的读写分离主要是在schema.xml文件中进行配置的，通过dataHost的3个属性结合其子标签和配 合完成 balance 属性负载均衡类型，目前的取值有 4 种： balance="0"，不开启读写分离机制，所有读操作都发送到当前可用的writeHost 上。 balance="1"，全部的 readHost 与 stand by writeHost 参与 select 语句的负载均衡，简单的说，当双 主双从模式(M1 ->S1 ， M2->S2，并且 M1 与 M2 互为主备)， 正常情况下， M2,S1,S2 都参与 select 语句的负载均衡。 balance="2"，所有读操作都随机的在 writeHost、 readhost 上分发。 balance="3"，所有读请求随机的分发到 wiriterHost 对应的 readhost 执行,writerHost 不负担读压 力，注意 balance=3 只在 1.4 及其以后版本有， 1.3 没有。 writeType 属性，负载均衡类型，目前的取值有 3 种： writeType="0"，所有写操作发送到配置的第一个 writeHost，第一个挂了切到还生存的第二个 writeHost，重新启动后已切换后的为准，切换记录在配置文件中:dnindex.properties . writeType="1"，所有写操作都随机的发送到配置的 writeHost，1.5 以后废弃不推荐。 writeType="2"，官方文档没有介绍。 <user name="root" > <property name="password">1234</property> <property name="schemas">db_train</property> <property name="defaultSchema">db_train</property> </user> 1 2 3 4 5 switchType 属性： -1 表示不自动切换 1 默认值，自动切换 2 基于MySQL 主从同步的状态决定是否切换 3、重启mycat进行测试，测试思路如下所示： 删除mysql11上的id为1的数据，然后在进行查询，如果此时查询不到数据，那么说明查询是从 mysql11（slave）节点进行查询的。如果可以查询到数据那么就说明查询还是使用的 mysql10（master）节点。 <!--配置逻辑库 --> <schema name="db_train" checkSQLschema="true" sqlMaxLimit="100" dataNode= "dataNode1"></schema> <!--配置分片节点 --> <dataNode name="dataNode1" dataHost="train_mysql_01" database="db_train" /> <!--配置节点主机--> <dataHost name="train_mysql_01" maxCon="1000" minCon="10" balance="1" writeType="0" dbType="mysql" dbDriver="native" switchType="1" slaveThreshold="100"> <heartbeat>select user()</heartbeat> <writeHost host="M1" url="172.19.0.110:3306" user="root" password="1234"> <readHost host="S1" url="172.19.0.111:3306" user="root" password="1234"/> </writeHost> </dataHost> 1 2 3 4 5 6 7 8 9 10 11 12 13 2.4.5 双主双从介绍 双主双部署 接下来我们就来给大家介绍一个mysql多主多从架构的主从复制，我们本次以双主双从为例来给大家演示 （为了不影响本次演示，我们最好删除之前一主一从的测试数据）。 主要分为两步进行实现： 1、双主双从的数据库配置实现 2、MyCat双主双从实现读写分离配置 MySQL环境部署 1、在指定的位置按照如下的目录结构创建目录（后期做MySql的目录映射） 2、将docker中的mysql配置文件分别复制到mysql20/confifig和mysql21/confifig目录中 3、对两个节点的配置文件做如下配置： /usr/local/mysql-master-slave/mysql20/confifig/mysql.conf.d mysql-master-slave mysql20 config data mysql21 config data 1 2 3 4 5 6 7 /usr/local/mysql-master-slave/mysql21/confifig/mysql.conf.d 4、更改mysql10节点的配置文件，添加如下配置： 5、创建mysql容器 lower_case_table_names=1 # 不区分字母大小写 log-bin=mysql-bin # 开启二进制日志 server-id=20 # 设置server-id log-slave-updates # 在作为从数据库的时候，有写入操作也要更新二进 制日志文件 # 不同步哪些数据库 binlog-ignore-db = mysql binlog-ignore-db = test binlog-ignore-db = information_schema 1 2 3 4 5 6 7 8 9 10 lower_case_table_names=1 # 不区分字母大小写 log-bin=mysql-bin # 开启二进制日志 server-id=21 # 设置server-id # 不同步哪些数据库 binlog-ignore-db = mysql binlog-ignore-db = test binlog-ignore-db = information_schema 1 2 3 4 5 6 7 8 9 log-slave-updates # 在作为从数据库的时候，有写入操作也要更新二进 制日志文件 重启容器 1 2 docker run -di --network=docker-network --ip=172.19.0.120 -- name=train_mysql_03 -p 3420:3306 --privileged=true -e MYSQL_ROOT_PASSWORD=1234 -v /usr/local/mysql-master slave/mysql20/data:/var/lib/mysql -v /usr/local/mysql-master slave/mysql20/config:/etc/mysql/ mysql:5.6 docker run -di --network=docker-network --ip=172.19.0.121 -- name=train_mysql_04 -p 3421:3306 --privileged=true -e MYSQL_ROOT_PASSWORD=1234 -v /usr/local/mysql-master slave/mysql21/data:/var/lib/mysql -v /usr/local/mysql-master slave/mysql21/config:/etc/mysql/ mysql:5.6 1 2 双主双从配置 1、进入到train_mysql_01容器中，查看该节点的mysql的主节点状态 其中File和Position都是我们在设置从（slave）数据库的时候用到的； 2、配置train_mysql_01的从数据库，需要配置的是train_mysql_02和train_mysql_03的数据库，配置如 下： train_mysql_02配置步骤如下所示： 3、配置train_mysql_03的从数据库，需要配置的是train_mysql_01和train_mysql_04的数据库，配置如 下： 进入到train_mysql_03的docker容器，查看train_mysql_03的主节点信息： mysql> show master status ; +------------------+----------+--------------+-------------------------- -----+-------------------+ | File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set | +------------------+----------+--------------+-------------------------- -----+-------------------+ | mysql-bin.000004 | 1799 | | mysql,test,information_schema | | +------------------+----------+--------------+-------------------------- -----+-------------------+ 1 row in set (0.00 sec) mysql> 1 2 3 4 5 6 7 8 9 stop slave;mysql> stop slave ; Query OK, 0 rows affected (0.00 sec) mysql> change master to master_host='172.19.0.110', master_user='root', master_password='1234', master_log_file='mysql-bin.000004', master_log_pos=1799; Query OK, 0 rows affected, 2 warnings (0.05 sec) mysql> start slave; Query OK, 0 rows affected (0.00 sec) mysql> show slave status\G 1 2 3 4 5 6 7 需要配置的是train_mysql_01和train_mysql_04的数据库，配置如下： mysql> show master status ; +------------------+----------+--------------+-------------------------- -----+-------------------+ | File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set | +------------------+----------+--------------+-------------------------- -----+-------------------+ | mysql-bin.000004 | 120 | | mysql,test,information_schema | | +------------------+----------+--------------+-------------------------- -----+-------------------+ 1 row in set (0.00 sec) 1 2 3 4 5 6 7 mysql> stop slave ; Query OK, 0 rows affected, 1 warning (0.00 sec) mysql> change master to master_host='172.19.0.120', master_user='root', master_password='1234', master_log_file='mysql-bin.000004', master_log_pos=120; Query OK, 0 rows affected, 2 warnings (0.06 sec) mysql> start slave ; Query OK, 0 rows affected (0.01 sec) 1 2 3 4 5 6 双主双从测试 接下来我们就来对刚才所部署的多主多从架构的数据库进行测试。具体的测试步骤如下所示： 1、使用Navicat分别连接train_mysql_01 , train_mysql_02,train_mysql_03,train_mysql_04节点。然后在 train_mysql_01 节点上创建一个数据库，并且创建表并且插入数据。 看看其他数据库是否存在变化。 2、在train_mysql_03 节点上创建一个数据库，然后在创建表并且插入数据。看看其他数据库是否存在变 化。 MyCat读写分离 接下来我们就来研究一下基于双主双从架构的mysql，mycat如何进行配置以实现双主的主备写入操作。具 体的实现步骤如下所示： 1、修改schema.xml文件train_mysql_01节点主机的配置 2、重启mycat 3、进行测试 由于我们之前已经测试过读写分离了，因此本次我们就不在测试读写分离了。本次我们主要测试两部分内 容： 1、读数据的负载均衡 2、写数据的高可用 读数据的负载均衡具体的测试步骤如下所示： <!--配置节点主机--> <dataHost name="train_mysql_01" maxCon="1000" minCon="10" balance="1" writeType="0" dbType="mysql" dbDriver="native" switchType="1" slaveThreshold="100"> <heartbeat>select user()</heartbeat> <writeHost host="M1" url="172.19.0.110:3306" user="root" password="1234"> <readHost host="S1" url="172.19.0.111:3306" user="root" password="1234"/> </writeHost> <writeHost host="M2" url="172.19.0.120:3306" user="root" password="1234"> <readHost host="S2" url="172.19.0.121:3306" user="root" password="1234"/> </writeHost> </dataHost> 1 2 3 4 5 6 7 8 9 10 1、创建数据库表tb_user 2、插入测试数据 此时我们发现4个节点上的数据都是相同的。 3、删除节点3上的4和5的数据，保留id为1,2,3的数据 4、删除节点4上的id为1,2,3的数据，重新插入id为4和5的数据 5、给节点2插入数据 6、通过mycat进行插入，观察数据的变化 最后有一点要说的是，在真实的项目中，不应该对从数据库（slave）做写入操作，这样会破坏数据的 一致性的。 create table tb_user( id int not null primary key , name varchar(20) , city varchar (20) ) ; 1 2 3 4 5 insert into tb_user (id, name , city) values (1 , "test1" , 'beijing') ; insert into tb_user (id, name , city) values (2 , "test2", 'xian') ; insert into tb_user (id, name , city) values (3 , "test3", 'shanghai') ; insert into tb_user (id , name , city) values (4 , "test4", 'shandong') ; insert into tb_user (id , name , city) values (5 , "test5", 'guangxi') ; 1 2 3 4 5 insert into tb_user (id , name , city) values (4 , "test4", 'shandong') ; insert into tb_user (id , name , city) values (5 , "test5", 'guangxi') ; 1 2 insert into tb_user (id , name , city) values (6 , "test6", 'guangxi') ; 1 写数据的高可用具体的测试步骤如下所示： 1、关闭train_mysql_01节点 2、通过mycat插入数据，观察数据是否可以被正常插入 3、关闭train_mysql_03节点，然后在尝试插入数据，观察数据是否可以被正常插入 insert into tb_user (id , name , city) values (7 , "test7", 'beijing') ; 1 2.4.6 订单数据库部署 接下来我们就需要按照我们之前所讲解的订单数据库的架构，来部署我们的订单数据库。 按照上述架构我们需要部署两套双主双从的mysql，双主双从的mysql我们已经讲解过了。因此第二套双主 双从的步骤我们不在课程中讲解了。 我们主要讲解的是mycat的配置，具体配置如下所示： 1、修改schema.xml文件 <!--配置逻辑库--> <schema name="db_train" checkSQLschema="true" sqlMaxLimit="100" dataNode="dataNode1"> <!-- 订单数据表分片 --> <table name="tb_order" dataNode="dataNode1,dataNode2" rule="sharding-by-murmur-order" primaryKey="id" /> </schema> <!--配置分片节点--> 1 2 3 4 5 6 7 8 9 2、在rule.xml加入订单数据分片规则 <dataNode name="dataNode1" dataHost="train_mysql_01" database="db_train" /> <dataNode name="dataNode2" dataHost="train_mysql_02" database="db_train" /> <!--配置节点主机 --> <dataHost name = "train_mysql_01" maxCon="1000" minCon="10" balance="1" writeType ="0" dbType = "mysql" dbDriver="native" switchType="1" slaveThreshold ="100" > < heartbeat > select user() </ heartbeat> <writeHost host ="M1" url= "172.19.0.110:3306" user="root" password="1234"> <readHost host="S1" url="172.19.0.111:3306" user="root" password="1234"/> </writeHost> <writeHost host="M2" url="172.19.0.120:3306" user="root" password="1234"> <readHost host="S2" url="172.19.0.121:3306" user="root" password="1234"/> </writeHost> </dataHost> <!--配置节点主机--> <dataHost name="train_mysql_02" maxCon="1000" minCon="10" balance="1" writeType="0" dbType="mysql" dbDriver="native" switchType="1" slaveThreshold="100"> <heartbeat>select user()</heartbeat> <writeHost host="M1" url="172.19.0.130:3306" user="root" password="1234"> <readHost host="S1" url="172.19.0.131:3306" user="root" password="1234"/> </writeHost> <writeHost host="M2" url="172.19.0.140:3306" user="root" password="1234"> <readHost host="S2" url="172.19.0.141:3306" user="root" password="1234"/> </writeHost> </dataHost> 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 <!-- 订单数据的分片规则 --> <tableRule name="sharding-by-murmur-order"> <rule> 1 2 3 3、连接mycat创建订单数据库表 <columns>id</columns> <algorithm>murmur-order</algorithm> </rule> </tableRule> <!-- 分片算法配置 --> <function name = "murmur-order" class="io.mycat.route.function.PartitionByMurmurHash"> <property name="seed">0</property> <property name = "count" >2</property> <!-- 定义分片数量 --> <property name = "virtualBucketTimes" > 160</ property > </function> 4 5 6 7 8 9 10 11 12 13 14 CREATE TABLE `tb_order` ( `id` bigint(25) NOT NULL, `riding_plan_date_id` bigint(25) DEFAULT NULL, `start_station_name` varchar(255) DEFAULT NULL, `receive_station_name` varchar(11) DEFAULT NULL, `train_num` varchar(11) DEFAULT NULL, `train_riding_date` varchar(20) DEFAULT NULL, `start_time` varchar(11) DEFAULT NULL, `end_time` varchar(11) DEFAULT NULL, `seat_num` varchar(10) DEFAULT NULL, `seat_nature` varchar(10) DEFAULT NULL, `coach_num` varchar(10) DEFAULT NULL, `seat_price` double(10,2) DEFAULT NULL, `pay_status` varchar(3) DEFAULT NULL, `user_id` varchar(100) DEFAULT NULL, `is_effect` varchar(3) DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 2.4.7 保存订单数据 接下来我们就来完成以下保存订单数据的代码，具体的实现步骤如下所示： 1、创建Mapper接口（TbOrderMapper） 2、创建Mapper映射文件（TbOrderMapper.xml） <!-- 保存订单数据 --> <insert id="saveOrder" parameterType="com.itheima.train.manager.domain.Order"> insert into tb_order (id , riding_plan_date_id , start_station_name , receive_station_name , train_num , train_riding_date , start_time , end_time ,seat_num , seat_nature , coach_num , seat_price , pay_status , user_id , is_effect) values ( #{id }, #{ridingPlanDateId } , 1 2 3 4 5 6 7 #{startStationName } , #{receiveStationName }, #{trainNum } , #{trainRidingDate } , #{startTime } , #{endTime }, #{seatNum }, #{seatNature } , #{coachNum} , #{seatPrice }, #{payStatus } , #{userId } , #{isEffect } ); </insert> <!-- 映射结果集 --> <resultMap id="BASE_RESULT_MAP" type="com.itheima.train.manager.domain.Order"> <id property="id" column="id" /> <result property="ridingPlanDateId" column="riding_plan_date_id" /> <result property="startStationName" column="start_station_name" /> <result property="receiveStationName" column="receive_station_name" /> <result property="trainNum" column="train_num"/> <result property="trainRidingDate" column="train_riding_date" /> <result property="startTime" column="start_time"/> <result property="endTime" column="end_time"/> <result property="seatNum" column="seat_num"/> <result property="seatNature" column="seat_nature"/> <result property="coachNum" column="coach_num"/> <result property="seatPrice" column="seat_price"/> <result property="payStatus" column="pay_status" /> <result property="isEffect" column="is_effect" /> <result property="userId" column="user_id" /> </resultMap> <!-- 根据用户的id ， 列车的编号，以及乘车日期查询订单信息 --> <select id="queryOrderInfo" parameterType="map" resultMap="BASE_RESULT_MAP"> select * from tb_order where user_id = #{userId} and train_num = # {trainNum} and train_riding_date = #{trainRidingDate} and is_effect = "1" 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 3、编写service，修改OrderHandlerConsumer类 2.4.8 删除排队信息 订单处理服务 当订单保存成功以后，我们就需要将用户的排队信息从Redis中进行删除。删除的具体流程如下所示： </select> 47 public Order saveOrder(Order order){ .... } 1 订单处理服务发送消息，具体的实现步骤如下所示： 1、添加队列配置信息 // 从Redis中删除排队信息的的队列 @Bean(name = "delete_sorted_from_redis") public Queue deleteSortedFromRedisQueue() { return QueueBuilder.durable("delete_sorted_from_redis").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBinddeleteSortedFromRedisQueue(@Qualifier(value = "train_manager_ex") Exchange exchange , @Qualifier(value = "delete_sorted_from_redis") Queue queue) { return BindingBuilder.bind(queue).to(exchange).with("delete_sorted").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 2、配置文件中添加支持生产者确认模式 3、发送消息OrderHandlerConsumer 订单保存成功以后，发送删除消息 订单服务 订单服务需要接收消息，将用户的排队信息从Redis中删除掉。具体的实现步骤如下所示： 1、配置文件中加入如下配置 publisher-confirm-type: correlated # 设置生产者通道支持confirm模式 1 // 调用service 进行数据保存 LOGGER.info("【保存订单数据完成】 ---> " + orderHandlerJsonInfo); order = orderService .saveOrder(orderHandler.getOrder()); // 发送删除Redis中排队信息的消息 String messageBody = order.getTrainNum() +":" + order.getTrainRidingDate() + ":" + order.getUserId() ; rabbitmqProducer.sendMessage(messageBody , "delete_sorted"); ... 1 2 3 4 5 6 7 8 9 2、添加队列配置信息 3、接收消息删除排队信息 listener: simple: acknowledge-mode: manual # 设置消息手动应答 1 2 3 // 从Redis中删除排队信息的的队列 @Bean(name = "delete_sorted_from_redis") public Queue deleteSortedFromRedisQueue () { return QueueBuilder.durable("delete_sorted_from_redis").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBinddeleteSortedFromRedisQueue(@Qualifier(value = "train_manager_ex") Exchange exchange , @Qualifier(value = "delete_sorted_from_redis") Queue queue) { return BindingBuilder.bind(queue).to(exchange).with("delete_sorted").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 @Component public class OrderConsumer { // 定义日志记录器 private static final Logger LOGGER = LoggerFactory.getLogger(OrderConsumer.class) ; // 依赖注入 @Autowired private StringRedisTemplate redisTemplate ; // 接收消息 @RabbitListener(queues = "delete_sorted_from_redis") public void deleteSortedFromRedis(String msg , Channel channel, Message message) { try { LOGGER.info("【删除排队信息开始】 ----> " + msg); 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 redisTemplate.boundZSetOps("train_manager:sorted_queue").remove(msg) ; LOGGER.info("【删除排队信息结束】 ----> " + msg); // 给出应答 channel.basicAck (message.getMessageProperties().getDeliveryTag() , false); } catch (IOException e) { e .printStackTrace (); LOGGER .info("【删除排队信息出错】 ----> " + msg); } } } 18 19 20 21 22 23 24 25 26 27 28 29 30 2.4.8 websocket应用 webSocket回顾 百度百科中针对webSocket的介绍如下： WebSocket 协议在2008年诞生，2011年成为国际标准，现在几乎所有浏览器都已经支持了。它的最大特点 就是，服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息，浏览器 和服务器只需要完成一次握手，两者之间就可以创建持久性的连接，并进行双向数据传输，也就是所谓的全 双工通讯的协议属于服务器推送技术的一种。 初次接触 WebSocket 的人，都会问同样的问题：我们已经有了 HTTP 协议，为什么还需要另一个协议？它 能带来什么好处？ 答案很简单: 因为 HTTP 协议有一个缺陷：通信只能由客户端发起，HTTP 协议做不到服务器主动向客户端推 送信息。 比如：没有websocket出现前我们获取数据是这样的 1、首先是 ajax轮询 ，ajax轮询 的原理非常简单，让浏览器隔个几秒就发送一次请求，询问服务器是否有新 信息 客户端：有没有新信息(Request) 服务端：没有（Response） 客户端：有没有新信息(Request) 服务端：没 有。。（Response） 客户端：有没有新信息 (Request) 服务端：没有 .......... 2、long poll（长轮询） 其实原理跟 ajax轮询 差不多，都是采用轮询的方式，不过采取的是阻塞模型，也就 是说，客户端发起连接后，如果没消息，就一直不返回 Response给客户端。直到有 消息才返回，返回完之后，客户端再次建立连接，周而复始。 从上面可以看出其实这两种方式，都是在不断地建立 HTTP连接，然后等待服务端处理，可以体现HTTP协议 的另外一个特点，被动性。何为被动性呢，其实就是，服务端不能主动联系客户端，只 能有客户端发起。 从上面很容易看出来，不管怎么样，上面这两种都是非常消耗资源的。 websocket通信 1、发送连接请求 客户端通过一个格式为：ws://host:port/的请求地址发起WebSocket连接请求，并由JavaScript实现 WebSocket API与服务器建立WebSocket连接，其中host为服务器主机IP地址或域名，port为端口。 2、握手 当服务器收到请求后，会解析请求头信息，根据升级后的协议判断该请求为WebSocket请求，并取出请求信 息中的Sec-WebSocket-Ke y字段的数据按照某种算法重新生成一个新的字符串序列放入响应 头Sec-WebSocket-Accept 中。 Sec-WebSocket-Accept ：服务器接受客户端HTTP协议升级的证明 3、WebSocket建立连接客户端接收服务器的响应后，同样会解析请求头信息，取出请求信息中的Sec WebSocket-Accept字段，并用服务器内部处理Sec-WebSocket-Key字段的算法处理之前发送的 Sec-WebSocket-Key，把处理得到的结果与Sec-WebSocket-Accept进行对比，数据相同则表示客户端与服务 器成功建立WebSocket连接，反之失败 服务端代码实现 具体的实现步骤，如下所示: 1、引入起步依赖 2、开启WebSocket支持端点 3、创建server核心类 因为WebSocket是类似客户端服务端的形式（采用ws协议），那么这里的WebSocketServer其实就相当于一 个ws协议的Server直接通过@ServerEndpoint("/im/{userId}") 、@Component启用即可， 然后在里面实现@OnOpen开启连接，@onClose关闭连接，@onMessage接收消息等方法。 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency> 1 2 3 4 /** * @Class: WebSocketConfig * @Package com.itheima.websocket.config * @Description: 开启WebSocket支持端点 * @Company: http://www.itheima.com/ */ @Configuration public class WebSocketConfig { @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 @Component @ServerEndpoint(value = "/im/{userId}") public class WebSocketServer { // 定义日志记录器 private Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class) ; // 创建一个Map集合，用来存储和服务端建立连接的会话信息 private static HashMap<String , Session> sessionHashMap = new HashMap<>() ; // 记录用户id private String userId = null ; 1 2 3 4 5 6 7 8 9 10 11 12 13 4、修改OrderHandlerConsumer类 保存完订单数据以后，调用WebSocketServer中的sendMessageToUser方法完成消息推送。 // 定义一个建立连接的方法 @OnOpen public void onOpen(Session session , @PathParam(value = "userId") String userId) { // 判断是否是已经存在该用户的连接 this .userId = userId ; if (sessionHashMap .containsKey (userId )) { sessionHashMap .remove(userId ) ; } sessionHashMap.put(userId , session) ; } @OnClose public void onClose() { if (sessionHashMap.containsKey(userId)) { sessionHashMap.remove(userId); } } // 推送消息 public void sendMessageToUser(String userId , String message) { try { // 获取客户端的session信息,发送消息 Session session = sessionHashMap.get(userId); session.getBasicRemote().sendText(message); } catch (IOException e) { e.printStackTrace(); } } } 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 5、Nginx配置虚拟主机 // 发送消息到 public void sendOrderInfo(String userId , boolean result , String orderId) { HashMap<String , Object> resultHashMap = new HashMap<>() ; resultHashMap.put("result" , result) ; resultHashMap .put ("orderId" , orderId ) ; webSocketServer .sendMessageToUser (userId , JSON.toJSONString (resultHashMap)); } 1 2 3 4 5 6 7 # 订单处理服务虚拟主机 server { listen 80; server_name www.trainmanager.order.handler.com; access_log logs/host.access.order.handler.log main ; # nginx访问日志 location / { # limit_req zone=rateLimit burst=5 nodelay; # limit_conn perip 1 ; # limit_conn perserver 100 ; proxy_pass http://train-manager-order-handler ; proxy_http_version 1.1; proxy_read_timeout 6000 ; # 设置超过时间，客户端 和服务器建立连接了以后，默认超过60s没有数据交互，nginx会自动将连接断开 proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; } error_page 500 502 503 504 /50x.html; location = /50x.html { root html; } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 2.4.9 查询订单接口 查询订单架构介绍 通过分析前端页面，我们可以看到在进行订单查询的时候调用的其实是itheima-train-manager-order项目中 的接口。而我们真正的订单信息是存储在数据库中的，而和数据库打交道的是 itheima-train-manager-order-handler工程，因此我们需要在itheima-train-manager-order-handler工程 中提供一个查询接口，供我们的itheima-train-manager-order来进行调用。 最初的调用架构： 弊端：订单处理服务进行扩容或者缩容的时候，我们都需要对订单服务做相应的更改 优化以后的调用架构： 本次我们还是选用nacos作为我们的服务注册中心。 服务提供方 集成Nacos 服务提供方集成nacos的具体步骤如下所示： 1、在pom.xml文件中添加如下依赖 2、在nacos配置中心中添加如下配置信息 3、在启动类上添加如下注解开启服务注册于发现功能 4、启动服务，就可以在nacos管理后台查看到对应的服务信息 <!-- nacos注册中心依赖 --> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos discovery</artifactId> </dependency> 1 2 3 4 5 spring: cloud: nacos: discovery: server-addr: 172.17.0.224:8848,172.17.0.224:8849,172.17.0.224:8850 1 2 3 4 5 @EnableDiscoveryClient 1 定义查询接口 具体步骤如下所示： 1、定义OrderHandlerController 2、OrderService定义查询方法 3、TbOrderMapper定义根据id查询方法 4、使用postman进行测试 @RestController @RequestMapping (value = "/orderHandler") public class OrderHandlerController { @Autowired private OrderService orderService ; @RequestMapping(value = "/queryById/{orderId}") public Order queryById(@PathVariable(value = "orderId") Long orderId) { return orderService.queryByOrderId(orderId) ; } } 1 2 3 4 5 6 7 8 9 10 11 12 13 // 根据订单的id进行查询 public Order queryByOrderId(Long orderId) { return tbOrderMapper.queryByOrderId(orderId); } 1 2 3 4 // 根据订单的id进行查询 public abstract Order queryByOrderId(Long orderId); 1 2 <!-- 根据订单的id查询订单信息 --> <select id="queryByOrderId" parameterType="java.lang.Long" resultMap="BASE_RESULT_MAP"> select * from tb_order where id = #{value} </select> 1 2 3 4 服务调用方 集成Nacos 服务调用方集成Nacos和服务提供方集成nacos方式一样，此处不再累述。 远程调用 1、定义根据id查询的接口 2、OrderService查询方法定义 @RequestMapping(value = "/queryById/{orderId}") public OrderHandler queryById(@PathVariable(value = "orderId") Long orderId) { return orderService.queryById(orderId); } 1 2 3 4 public OrderHandler queryById(Long orderId) { // TODO 远程调用itheima-train-manager-order-handler查询订单信息接口 Order order = null ; // 从Redis中获取用户信息 User user = (User) redisTemplate.boundHashOps("train_manager:userInfo").get(order.getUserI d()) ; 1 2 3 4 5 6 7 8 3、进行远程调用 Feign基本使用 由于itheima-train-manager-order-handler提供的是一个http接口。因此要调用该接口，我们就需要使用一 些Http的客户端工具。比如HttpClient ， RestTemplate等。要想进行负载 均衡的调用，我们可以使用Spring Cloud中的Ribbon组件实现。本次我们在进行远程调用的时候，我们将会 使用Spring Cloud中的另外一个组件：Feign Feign是一个声明式的伪RPC的REST客户端，它用了基于接口的注解方式，很方便实现客户端配置。将Java Http 客户端绑定到它的内部，并且在内部集成了Ribbon组件， Feign 的首 要目标是将Java Http客户端调用过程变得简单。 具体的使用步骤如下所示： 1、在pom.xml文件中添加如下依赖 2、定义远程调用接口 3、启动类上添加如下注解，开启Feign的支持 // 创建OrderHandler对象 OrderHandler orderHandler = new OrderHandler() ; orderHandler.setOrder(order); orderHandler.setUser(user); // 返回 return orderHandler ; } 9 10 11 12 13 14 15 16 17 <!--Feign依赖加入--> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-openfeign</artifactId> </dependency> 1 2 3 4 5 @FeignClient(name = "itheima-train-manager-order-handler") public interface OrderFeignClient { @RequestMapping(value = "/order/queryById/{orderId}") public abstract Order queryById(@PathVariable(value = "orderId") Long orderId) ; } 1 2 3 4 5 6 7 4、进行远程调用 @EnableFeignClients 1 @Service public class OrderService { @Autowired private OrderFeignClient orderFeignClient ; public OrderHandler queryById(Long orderId) { // TODO 远程调用itheima-train-manager-order-handler查询订单信息接口 Order order = orderFeignClient.queryById(orderId) ; ... // 返回 return orderHandler ; } } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 排队方案测试 为了看到明显的测试效果，我们可以在订单处理服务中加入线程休眠时间，模拟业务的处理时长。 更改页面的用户id开启多个页面进行测试。 2.5 下单优化 2.5.1 问题描述 现在我们的下单看似已经完成了，但是还是存在一些小问题，比如： 1、如果该用户已经下单了，我们还需要进行预扣库存操作吗？ 2、用户虽然已经下单了，但是一直没有对订单进行支付。 3、进行预扣减库存的时候存在线程安全问题。 2.5.2 预扣库存优化 当该用户已经购买了指定列车的火车票，那么我们就不能再进行预扣库存操作了。 实现思路如下所示： 1、在订单处理服务中提供一个查询接口（根据用户的 id ， 已经乘车计划id），查询订单信息 2、在订单服务中，在进行预扣库存之前远程调用接口，查询订单信息 3、修改前端页面针对结果是false的处理情况 订单处理服务 查询接口如下所示： OrderHandlerController添加查询接口 OrderService添加查询方法 @RequestMapping(value = "/queryOrderInfo/{userId}/{trainNum}/{trainRidingDate}") public Order queryOrderInfo(@PathVariable(value = "userId") String userId , @PathVariable(value = "trainNum") String trainNum , @PathVariable(value = "trainRidingDate") String trainRidingDate) { return orderService.queryOrderInfo(userId , trainNum , trainRidingDate) ; } 1 2 3 4 订单服务远程调用 OrderFeignClient添加查询接口 OrderService添加远程调用逻辑 // 查询订单信息 public Order queryOrderInfo(String userId, String trainNum, String trainRidingDate) { // 构建查询条件 HashMap< String , Object > hashMap = new HashMap<>() ; hashMap. put("userId" , userId ) ; hashMap . put ( "trainNum" , trainNum ) ; hashMap . put ( "trainRidingDate" , trainRidingDate ) ; Order dbOrder = tbOrderMapper.queryOrderInfo(hashMap); // 返回 return dbOrder ; } 1 2 3 4 5 6 7 8 9 10 11 12 13 @RequestMapping(value = "/orderHandler/queryOrderInfo/{userId}/{trainNum}/{trainRidingDate}") public abstract Order queryOrderInfo(@PathVariable(value = "userId") String userId , @PathVariable(value = "trainNum") String trainNum , @PathVariable(value = "trainRidingDate") String trainRidingDate) ; 1 2 ... // 发起远程调用 Order dbOrder = orderFeignClient.queryOrderInfo(orderParams.getUserId(), trainNum, trainRidingDate); if(dbOrder != null) { responseResult.setResult(false); responseResult.setData(String.valueOf(dbOrder.getId())); responseResult.setMessage("1"); // 表示订单信息已经存在了 return responseResult ; } // 构建乘车计划的key // riding_plan:C1001:2020-06-01:264546196628963328 LOGGER.info("【预扣库存开始了】 ---> " + JSON.toJSONString(orderParams)); String ridingPlanDateKey = "riding_plan:" + trainNum + ":" + trainRidingDate + ":" + ridingPlanId ; 1 2 3 4 5 6 7 8 9 10 11 12 13 14 前端页面修改 int count = Integer.parseInt(redisTemplate.boundHashOps(ridingPlanDateKey).get(redi sticketTypeKey).toString()) ; if(count <= 0 ) { responseResult.setResult(false); responseResult .setMessage ("0"); // 表示票已经售罄了 return responseResult ; } ... 15 16 17 18 19 20 21 axios.post("http://www.trainmanager.order.com/order/submitOrder" , this.orderParams).then(function(response){ if(response.data.result) { // 下单排队成功，跳转到订单查询页面 window.location.href="http://www.trainmanager.com/otn/payOrder/preOrder Queue.html?ridingPlanId=" + _this.orderParams.ridingPlanId + "&trainNum=" + _this.orderParams.trainNum + "&trainRidingDate=" + _this.orderParams.trainRidingDate + "&userId=" + _this.orderParams.userId + "&ticketType=" + _this.orderParams.ticketType; }else { if(response.data.message == "1") { window.location.href = "http://www.trainmanager.com/otn/payOrder/init.html?orderId=" + response.data.data ; }else if(response.data.message == "0"){ window.location.href="/otn/payOrder/genOrderFail.html" ; } } }) ; 1 2 3 4 5 6 7 8 9 10 11 12 2.5.3 回退库存 延迟队列 延迟队列存储的对象是对应的延迟消息，所谓"延迟消息"是指当消息被发送以后，并不想让消费者立刻拿到 消息，而是等待特定时间后，消费者才能 拿到这个消息进行消费。而我们的在完成订单未支付超时这个需求的时候就可以去使用延迟队列。那么我们 首先来给大家介绍一下具体的实现流程， 如下图所示： 死信队列 在Rabbitmq中没有延迟队列的功能，但是我们可以使用DLX结合TTL来实现延迟队列的功能。 DLX：可以被称之为死信交换机（Dead-Letter-Exchange）,也有人将其称之为死信邮箱。当消息在一个队列 中变成死信之后，这个消息能被重新发送到另外一个交换机，这个交换机就是DLX，与之 绑定的队列就是死信队列。 一个消息成为死信的情况： 1、消息被拒绝，并且设置requeue参数为false 2、消息过期 3、队列达到最大长度 如下下图所示： 发送延迟消息 接下来我们就来演示一下使用死信队列模拟延迟队列。 订单服务 1、在（itheima-train-manager-or der）RabbitmqConfifiguration中添加死信队列的与死信交换机的配置 2、在RabbitmqConfifiguration中添加延迟队列信息 // 声明回退库存的死信交换机 @Bean(name = "order_timeout_dlx_ex") public Exchange backStockDlxExchange() { return ExchangeBuilder.directExchange("order_timeout_dlx_ex").durable(true).bu ild() ; } // 声明回退库存的死信队列 @Bean(name = "order_timeout_dlx_queue") public Queue backStockDlxQueue() { return QueueBuilder.durable("order_timeout_dlx_queue").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding backStockDlxExBindBackStockDlxQueue(@Qualifier(value = "order_timeout_dlx_ex") Exchange exchange , @Qualifier(value = "order_timeout_dlx_queue") Queue queue) { return BindingBuilder.bind(queue).to(exchange).with("order_timeout").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 // 声明回退库存的队列 @Bean(name = "order_timeout_queue") public Queue orderTimeOutQueue() { QueueBuilder queueBuilder = QueueBuilder.durable("order_timeout_queue"); // 订单队列建造者对象 queueBuilder.ttl(1000 * 60); // 单位为毫秒 1 2 3 4 5 3、发送订单延迟消息 订单处理服务 1、在（itheima-train-manager-order-handler）RabbitmqConfifiguration按照上述配置信息添加同样的配 置 2、在OrderHandlerConsumer类中添加监听队列方法 queueBuilder.deadLetterExchange("order_timeout_dlx_ex") ; // 设置死信交换机 queueBuilder.deadLetterRoutingKey("order_timeout") ; // 死信交换机路由键 return queueBuilder.build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBindOrderTimeOutQueue (@Qualifier (value = "train_manager_ex" ) Exchange exchange , @Qualifier (value = "order_timeout_queue") Queue queue) { return BindingBuilder.bind(queue).to(exchange).with("order_timeout").noargs() ; } 6 7 8 9 10 11 12 13 14 15 // 发送订单数据 rabbitmqProducer.sendMessage(JSON.toJSONString(orderHandler) , "gen_order"); // 发送延迟消息到MQ中 LOGGER.info("【发送延迟消息开始了】 ---> " + JSON.toJSONString(order)); HashMap<String , Long> msg = new HashMap<>() ; msg.put("orderId" , order.getId()) ; rabbitmqProducer.sendMessage(JSON.toJSONString(msg) , "order_timeout"); LOGGER.info("【发送延迟消息结束了】 ---> " + JSON.toJSONString(order)); 1 2 3 4 5 6 7 8 9 @RabbitListener(queues = "order_timeout_dlx_queue") public void orderTime(String orderTimeOutJson , Channel channel , Message message) { try { LOGGER.info("【获取到订单超时消息】 ---> " + orderTimeOutMsg); 1 2 3 4 5 6 生产者代码 生产者代码实现思路： 1、根据id查询订单数据 2、判断订单的支付状态 3、如果是未支付，修改订单的有效状态为为无效 4、并且向mq中发送回退库存消息 具体是实现步骤如下所示： HashMap<String , Long> msg = JSON.parseObject(orderTimeOutMsg , HashMap.class) ; Long orderId = msg.get("orderId"); // 查询数据库 // 给出应答 channel. basicAck(message.getMessageProperties().getDeliveryTag() , false); } catch (IOException e) { e.printStackTrace(); } } 7 8 9 10 11 12 13 14 15 16 17 18 19 1、定义根据id修改订单信息方法 2、Mapper映射文件内容 3、OrderService代码 4、OrderHandlerConsumer中orderTime修改 // 修改订单信息 public abstract void updateOrderById(Order order) ; 1 2 <!-- 根据 id 进行修改 --> <select id = "updateOrderById" parameterType="com.itheima.train.manager.domain.Order"> update tb_order set riding_plan_date_id = #{ridingPlanDateId} , start_station_name = #{startStationName} , receive_station_name = #{receiveStationName}, train_num = #{trainNum} , train_riding_date = #{trainRidingDate} , start_time = #{startTime} , end_time = #{endTime} , seat_num = #{seatNum} , seat_nature = #{seatNature} , coach_num = #{coachNum} , seat_price = #{seatPrice} , pay_status = #{payStatus} , user_id = #{userId } , is_effect = #{isEffect} where id = #{id} </select> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 // 修改订单信息 public void updateOrderById(Order order) { orderMapper.updateOrderById(order); } 1 2 3 4 @RabbitListener(queues = "order_timeout_dlx_queue") public void orderTime(String orderTimeOutJson , Channel channel , Message message) { try { 1 2 3 4 5 5、在RabbitmqConfifiguration添加回退库存队列信息 回退库存的队列信息如下图所示： // 解析数据 HashMap hashMap = JSON.parseObject(orderTimeOutJson, HashMap.class); Long orderId = Long.parseLong(hashMap.get("orderId").toString()); // 调用 service查询订单信息 Order order = orderService.queryByOrderId(orderId); // 判断订单支付状态 if( "0".equals(order.getPayStatus())) { // 修改订单的有效性 order.setIsEffect("0"); // 更新数据 orderService.updateOrderById(order); // TODO发送订单消息 logger.info("【订单超时消息成功获取，回退库存成功】--->" + orderTimeOutJson); } channel.basicAck(message.getMessageProperties().getDeliveryTag() , false); } catch (IOException e) { e.printStackTrace(); logger.info("【订单超时消息成功获取，回退库存是失败】--->" + orderTimeOutJson); } } 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 6、发送回退库存队列消息 // 声明队列 @Bean(name = "back_stock_redis_queue" ) public Queue backStockRedisQueue () { return QueueBuilder.durable("back_stock_redis_queue").build() ; } // 声明队列 @Bean(name = "back_stock_es_queue") public Queue backStockESQueue() { return QueueBuilder.durable("back_stock_es_queue").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBindBackStockRedisQueue(@Qualifier(value = "train_manager_ex") Exchange exchange , @Qualifier(value = "back_stock_redis_queue") Queue queue) { return BindingBuilder.bind(queue).to(exchange).with("back_stock").noargs() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBindBackStockESQueue(@Qualifier(value = "train_manager_ex") Exchange exchange , @Qualifier(value = "back_stock_es_queue") Queue queue) { return BindingBuilder.bind(queue).to(exchange).with("back_stock").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 // 判断订单支付状态 if("0".equals(order.getPayStatus())) { // 修改订单的有效性 order.setIsEffect("0"); 1 2 3 4 5 消费者代码 Redis库存回退 1、在itheima-train-manager-redis的RabbitmqConfifiguration添加回退库存队列信息 // 更新数据 orderService.updateOrderById(order); // TODO发送订单消息 rabbitmqProducer .sendMessage(JSON.toJSONString(order) , "back_stock" ); LOGGER .info ("【订单超时消息成功获取，回退库存消息发送成功】 ---> " + orderTimeOutMsg ); } 6 7 8 9 10 11 12 13 14 2、给RedisConsumer添加回退库存方法 // 声明队列 @Bean(name = "back_stock_redis_queue") public Queue backStockRedisQueue() { return QueueBuilder.durable("back_stock_redis_queue").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBindBackStockRedisQueue(@Qualifier(value = "train_manager_ex" ) Exchange exchange , @Qualifier(value = "back_stock_redis_queue" ) Queue queue ) { return BindingBuilder.bind(queue).to(exchange).with("back_stock").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 @Component public class RedisConsumer { // 定义日志记录器 private static final Logger LOGGER = LoggerFactory.getLogger(RedisConsumer.class) ; // 依赖注入 @Autowired private StringRedisTemplate redisTemplate ; @RabbitListener(queues = "back_stock_redis_queue") public void backStock(String orderJson , Channel channel , Message message) { try { // 解析数据 LOGGER.error("【Redis回退库存开始了】 ---> " + orderJson); Order order = JSON.parseObject(orderJson , Order.class) ; String stockCountKeyDynamic = "riding_plan:" + order.getTrainNum() + ":" + order.getTrainRidingDate() + ":*" ; Set<String> keys = redisTemplate.keys(stockCountKeyDynamic); for(String key : keys) { 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 BoundHashOperations<String, Object, Object> boundHashOperations = redisTemplate.boundHashOps(key); switch (order.getSeatNature()) { case "3" : int firstSeatStock = Integer.parseInt (boundHashOperations.get("firstSeatStock").toString()) ; boundHashOperations .put("firstSeatStock" , String.valueOf (firstSeatStock + 1)); break ; case "4" : int secondSeatStock = Integer.parseInt(boundHashOperations.get("secondSeatStock").toString()) ; boundHashOperations.put("secondSeatStock" , String.valueOf(secondSeatStock + 1)); break ; case "7" : int hardSleeperStock = Integer.parseInt(boundHashOperations.get("hardSleeperStock").toString() ) ; boundHashOperations.put("hardSleeperStock" , String.valueOf(hardSleeperStock + 1)); break ; case "6" : int softSleeperStock = Integer.parseInt(boundHashOperations.get("softSleeperStock").toString() ) ; boundHashOperations.put("softSleeperStock" , String.valueOf(softSleeperStock + 1)); break ; case "2" : int busSeatStock = Integer.parseInt(boundHashOperations.get("busSeatStock").toString()) ; boundHashOperations.put("busSeatStock" , String.valueOf(busSeatStock + 1)); break ; case "5" : int hardSeatStock = Integer.parseInt(boundHashOperations.get("hardSeatStock").toString()) ; boundHashOperations.put("hardSeatStock" , String.valueOf(hardSeatStock + 1)); break ; case "9" : 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 ES库存回退 1、在itheima-train-manager-es的RabbitmqConfifiguration添加回退库存队列信息 int softSeatStock = Integer.parseInt(boundHashOperations.get("softSeatStock").toString()) ; boundHashOperations.put("softSeatStock" , String.valueOf(softSeatStock + 1)); break ; } } LOGGER.info("【Redis回退库存成功了】 ---> " + orderJson); // 释放座位 LOGGER.info("【Redis释放座位开始了】 ---> " + orderJson); String seatNatureKey = "train_seat:" + order.getTrainNum() + ":" + order.getTrainRidingDate() + ":" + order.getSeatNature() + ":" + order.getCoachNum() ; redisTemplate.boundHashOps(seatNatureKey).put(order.getSeatNum() , "0"); LOGGER.info("【Redis释放座位成功了】 ---> " + orderJson); // 给出应答 channel.basicAck(message.getMessageProperties().getDeliveryTag() , false); } catch (IOException e) { e.printStackTrace(); LOGGER.error("【Redis回退库存失败】 ---> " + orderJson); } } } 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 2、给EsConsumer添加回退库存方法 // 声明队列 @Bean(name = "back_stock_es_queue") public Queue backStockESQueue() { return QueueBuilder.durable("back_stock_es_queue").build() ; } // 声明交换机和队列的绑定关系 @Bean public Binding trainExBindBackStockESQueue(@Qualifier(value = "train_manager_ex" ) Exchange exchange , @Qualifier(value = "back_stock_es_queue" ) Queue queue ) { return BindingBuilder.bind(queue).to(exchange).with("back_stock").noargs() ; } 1 2 3 4 5 6 7 8 9 10 11 // 回退库存 public void backStock(Order order) throws IOException { // 创建搜索请求对象 SearchRequest searchRequest = new SearchRequest("riding_plan_date") ; SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().trackTotalHits(true) ; // 表示要查询所有的数据 BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); boolQueryBuilder.must(QueryBuilders.termQuery("trainNum" , order.getTrainNum())) ; boolQueryBuilder.must(QueryBuilders.termQuery("trainRidingDate" , order.getTrainRidingDate())) ; searchSourceBuilder.query(boolQueryBuilder); searchRequest.source(searchSourceBuilder) ; // 进行搜索 SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); SearchHits searchHits = searchResponse.getHits(); SearchHit[] hitsHits = searchHits.getHits(); for (SearchHit hitsHit : hitsHits) { String esSearchResult = hitsHit.getSourceAsString(); TbRidingPlanDate ridingPlanDate = JSON.parseObject(esSearchResult, TbRidingPlanDate.class); switch (order.getSeatNature()) { 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 case "3" : ridingPlanDate.setFirstSeatStock(ridingPlanDate.getFirstSeatStock() + 1); break ; case "4": ridingPlanDate .setSecondSeatStock(ridingPlanDate.getSecondSeatStock() + 1); break ; case "7": ridingPlanDate.setHardSleeperStock(ridingPlanDate.getHardSleeperStock( ) + 1); break ; case "6": ridingPlanDate.setSoftSleeperStock(ridingPlanDate.getSoftSleeperStock( ) + 1); break; case "2": ridingPlanDate.setBusSeatStock(ridingPlanDate.getBusSeatStock() + 1); break ; case "5": ridingPlanDate.setHardSeatStock(ridingPlanDate.getHardSeatStock() + 1); break ; case "9": ridingPlanDate.setSoftSeatStock( ridingPlanDate.getSoftSeatStock() + 1); break ; } // 创建更新请求对象 UpdateRequest updateRequest = new UpdateRequest("riding_plan_date" , hitsHit.getId()).doc(JSON.toJSONString(ridingPlanDate) , XContentType.JSON) ; restHighLevelClient.update(updateRequest , RequestOptions.DEFAULT) ; } 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 2.5.4 分布式锁 分布式锁简介 现在我们的下单预扣减库存的代码还存在一个比较严重的问题-线程安全问题。 } 49 50 // 构建乘车计划的key LOGGER.info("【预扣库存开始了】 ---> " + JSON.toJSONString(orderParams)); String ridingPlanDateKey = "riding_plan:" + trainNum + ":" + trainRidingDate + ":" + ridingPlanId ; int count = Integer.parseInt(redisTemplate.boundHashOps(ridingPlanDateKey).get(redi sticketTypeKey).toString()) ; if(count <= 0 ) { responseResult.setResult(false); responseResult.setMessage("0"); // 表示票已经售罄了 return responseResult ; 1 2 3 4 5 6 7 8 在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者Lock的方式来解决多线程 间的代码同步问题，这时多线程的运行都是在同一个JVM之下。但当我们的应用是分布式集 群工作的情况下，属于多JVM下的工作环境，JVM之间已经无法通过多线程的锁解决同步问题。那么就需要 一种更加高级的锁机制，来处理种跨机器的进程之间的数据同步问题——这就是分布式锁。 分布式锁特性 首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件： 1、互斥性：任意时刻，只能有一个客户端获取锁，不能同时有两个客户端获取到锁。 2、安全性：锁只能被持有该锁的客户端删除，不能由其它客户端删除。 3、高可用性：：当部分节点（ redis节点等）down机时，客户端仍然能够获取锁和释放锁。 4、失效性：获取锁的客户端因为某些原因（如down机等）宕机，那么此时就需要释放锁，如果没有及时释 放锁，可能会导致死锁现象。 分布式锁实现方式 分布式锁的实现方式比较多： } ... // 动态库存扣减 String stockCountKeyDynamic = "riding_plan:" + trainNum + ":" + trainRidingDate + ":*" ; Set<String> keys = redisTemplate .keys(stockCountKeyDynamic); for(String key : keys ) { redisTemplate .boundHashOps (key).put(redisticketTypeKey , String.valueOf (count - 1)); } LOGGER.info("【预扣库存结束了】 ---> " + JSON.toJSONString(orderParams)); 9 10 11 12 13 14 15 16 17 18 本次我们重点介绍的就是使用zookeeper结合它的客户端API（Curator）实现分布式锁。数据模型 数据模型介绍 zookeeper的内部结构如下图所示： 从上图可以看到，ZooKeeper的数据模型和Unix的文件系统目录树很类似，拥有一个层次的命名空间。这里 面的每一个节点都被称为 - ZNode， 节点可以拥有子节点，同时也允许少 量数据节点存储在该节点之下。ZooKeeper 数据模型是由一系列基本数据单元 Znode (数据节点) 组成的节 点树，其中根节点为 / ，每个节点上都会保存自己的数据和节点信息。 ZooKeeper 中节点可以分为四大类，分别是： PERSISTENT 持久化节点 即使在创建该特定znode的客户端断开连接后，持久节点仍然存在。默认情况下，除非另有说明，否则所建 的znode都是持久的 PERSISTENT_SEQUENTIAL 顺序节点 当一个新的znode被创建为一个顺序节点时， ZooKeeper通过将10位的序列号附加到原始名称来设置znode 的路径。 例如，如果将具有路径 /myapp 的znode创建为顺序节点，则 ZooKeeper 会将路径更改为 /myapp0000000001 ，并将下一个序列号设置为 0000000002 。 如果两个顺序节点是同时创建的，那么ZooKeeper不会对每个znode使用相同的数字。顺序节点在锁定和同 步中起重要作用。 EPHEMERAL 临时节点 客户端session超时的时候该节点就会被自动删除，客户端活跃时，临时节点就是有效的。 EPHEMERAL_SEQUENTIAL 临时顺序节点 删除时机和临时节点一样，唯一不同的就是临时有序节点会在uuid后面加入有序的序列 数据模型测试 具体的步骤如下所示： 1、使用zookeeper的客户端工具连接到zookeeper上（为了方便我们可以使用windows版本的zookeeper对 应的客户端工具进行测试）（C:\develop\apache-zookeeper-3.5.7-bin） 2、创建节点（create [-s] [-e] path data acl） C:\develop\apache-zookeeper-3.5.7-bin\bin>zkCli.cmd -server 127.0.0.1:2181 1 3、查询节点数据 4、修改节点数据 5、删除节点 create /itheima01 itheima01 # 创建一个持久节点 create -s /ithema02 itheima02 # 创建一个持久有序节点 create -e /itheima03 itheima03 # 创建一个临时节点 create -s -e /itheima04 itheima04 # 创建一个临时有序节点 1 2 3 4 get path 1 set path data 1 delete path 1 事件监听器 基本介绍 ZooKeeper允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper服务端 会将事件通知到感兴趣的客户端上去，该机制是ZooKeeper实现分布式协调服务的重要特性。 ZooKeeper中引入了Watcher 机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一 个对象自身状态变化时，会通知所有订阅者。zookeeper的watcher是一次性的，触发后即 销毁。 watcher架构 Watcher实现由三个部分组成： Zookeeper服务端； Zookeeper客户端； 客户端的ZKWatchManager对象； 客户端首先将Watcher注册到服务端，同时将Watcher对象保存到客户端的Watch管理器中。当ZooKeeper 服务端监听的数据状态发生变化时，服务端会主动通知客户端， 接着客户端的Watch管理器会触发相关Watcher来回调相应处理逻辑，从而完成整体的数据发布/订阅流程。 watcher测试 1、在使用get命令获取节点数据的时候，我们可以对其添加一个watcher 2、通过另外一个客户端去更改节点数据 3、查看该客户端的变化 get -w /itheima01 1 set /itheima01 itcast01 1 WatchedEvent state:SyncConnected type:NodeDataChanged path:/itheima01 1 分布式锁原理 zookeeper实现分布式锁的原理如下图所示： 注意：判断自己是否是locks目录下序号最小的节点，只会和比自己序号小的那个相邻节点进行比较，这样大 大提高了效率。比如N节点只会和N-1几点比较，不会和N+1，N-2节点比较分布式锁实现 Curator简介 针对zookeeper实现分布式锁，我们没有必须在自己编写代码进行实现了。zookeeper的Java客户端Curator 已经提供了分布式锁的实现。 zookeeper的原生api相对来说比较繁琐，比如：对节点添加监听事件，当监听触发后，我们需要再次手动添 加监听，否则监听只生效一次；再比如，断线重连也需要我们手动代码来判断处理等。 Curator是Netflflix开源的一套ZooKeeper客户端框架，用它来操作zookeeper更加简单方便。 Curator框架提供了一套高级的API， 简化了ZooKeeper的操作。它增加了很多使用ZooKeeper开发的特 性，可以处理ZooKeeper集群复杂的连接管理和重试机制。 目前已经成为Apache的顶级项目。另外还提供了一套易用性和可读性更强的Fluent风格的客户端API框架。 下面是官网http://curator.apache.org/对Curator的描述： Curator技术栈 curator有很多模块，核心的模块为:curator-framework Client:封装原生ZooKeeper 类，管理和ZooKeeper集群的连接，并提供了重连的机制 Framework：Framework 是ZooKeeperClient更高的抽象API。 自动连接管理:当ZooKeeper客户端内部出现异常，将自动进行重连或重试，该过程对外几乎完全透明， 更清晰的API:简化了ZooKeeper原生的方法，事件等， 提供流程的接口 。 Recips：使用Framework实现了大量的ZooKeeper的协同服务,它封装了一些高级特性，如：Cache事件监 听、选举、分布式锁、分布式计数器、分布式Barrier等， 该组件建立在Framework的基础之上。 Extensions：扩展模块 分布式锁实现 在Curator中有五种锁方案，主要包含： InterProcessMutex：分布式可重入排它锁 InterProcessSemaphoreMutex：分布式排它锁（非可重入锁） InterProcessReadWriteLock：分布式读写锁 InterProcessMultiLock：将多个锁作为单个实体管理的容器 InterProcessSemaphoreV2 ：共享信号量 一般情况下我们使用的都是InterProcessMutex。 curator分布式锁实现 具体的实现步骤如下所示： 1、导入相关的依赖包 2、在nacos配置中心中添加zookeeper集群节点的配置 3、创建CuratorFramework对象 4、创建InterProcessMutex对象 <dependency > <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>4.0.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>4.0.0</version> </dependency> 1 2 3 4 5 6 7 8 9 10 11 zk: cluster: nodes: 127.0.0.1:2181,127.0.0.1:2281,127.0.0.1:2381 1 2 3 // 定义连接重试策略，第一个参数表示两次连接的等待时间，第二个参数表示最大的重试次数 ExponentialBackoffRetry exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3); // 获得CuratorFramework对象 CuratorFramework client = CuratorFrameworkFactory.builder().connectString(“zk集群的连接地址”) .retryPolicy(exponentialBackoffRetry) .namespace("itheima-train-znode") // 定义一个名称空间 .sessionTimeoutMs(5000) // 连接超时时间 .build(); client.start(); // 启动客户连接 1 2 3 4 5 6 7 8 9 10 5、调用InterProcessMutex想法方法获取锁和释放锁 InterProcessMutex lock = new InterProcessMutex(client , "/lock-znode") ; 1 // 开始获取锁，指定获得锁最大的等待的时间，抢夺时，如果出现堵塞，会在超过该时间后，返 回false public boolean acquire(long time, TimeUnit unit) public void release() // 释放锁 1 2 3
标签：String,public,1111,订单,train,mysql,order From： https://www.cnblogs.com/shan13936/p/17416602.html