。

以文章为主线的核心业务主要分为如下子模块。

自媒体模块实现用户创建功能、文章发布功能、素材管理功能。

app端用户模块实现文章搜索、文章点赞、…总体框架

本项目主要着手于获取最新最热新闻资讯以微服务构架为技术基础搭建校内仅供学生教师使用的校园新媒体app。

以文章为主线的核心业务主要分为如下子模块。

自媒体模块实现用户创建功能、文章发布功能、素材管理功能。

app端用户模块实现文章搜索、文章点赞、关注、商家优惠卷秒杀等功能。

业务可以帮助商家引流增加曝光度也可以为用户提供查看提供附近消费场所。

主要技术SpringBootSpringcloudMySQLMyBatisPlusRedisElasticSearchKafkaMongoDBxxl-job

主要职责

采用Redis作缓存并手动封装了一个工具类通过互斥锁逻辑过期的方式解决缓存击穿问题·。

1.1

Executors.newFixedThreadPool(10);//构造函数注入stringRedisTemp也可以用resource注解public

CacheClient(StringRedisTemplate

stringRedisTemplate)

{stringRedisTemplate.opsForValue().set(key,

time,

RedisData();redisData.setData(value);redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));//

写入RedisstringRedisTemplate.opsForValue().set(key,

JSONUtil.toJsonStr(redisData));}

//方法3

queryWithPassThrough(//前缀id返回值类型数据库查询的逻辑有参有返回值的函数ttlttl单位String

keyPrefix,

stringRedisTemplate.opsForValue().get(key);//

{//

将空值写入redisstringRedisTemplate.opsForValue().set(key,

CACHE_NULL_TTL,

queryWithLogicalExpire(//前缀id返回值类型数据库查询的逻辑有参有返回值的函数ttlttl单位String

keyPrefix,

stringRedisTemplate.opsForValue().get(key);//

{//

5.判断是否过期if(expireTime.isAfter(LocalDateTime.now()))

{//

6.3.成功开启独立线程实现缓存重建CACHE_REBUILD_EXECUTOR.submit(()

{try

方法2this.setWithLogicalExpire(key,

newR,

stringRedisTemplate.opsForValue().get(key);//

{//

4.3.获取锁失败休眠并重试Thread.sleep(50);return

id,

将空值写入redisstringRedisTemplate.opsForValue().set(key,

CACHE_NULL_TTL,

stringRedisTemplate.opsForValue().setIfAbsent(key,

10,

{stringRedisTemplate.delete(key);}

我们可以直接看方法4这里通过泛型的方式对输入的参数做了以下处理

2、存在需要先把json反序列化为对象

4、过期了尝试获取锁对象此处采用Redis中的setnx方法实现分布式锁的功能。

使用店铺前缀加商品id作为key

5、没有获取到返回过期数据。

获取到了使用线程池的submit方法开启新的线程去重建缓存也就是更新过期时间

6、重建完成释放锁

cacheClient.queryWithPassThrough(CACHE_SHOP_KEY,

id,

.queryWithMutex(CACHE_SHOP_KEY,

id,

.queryWithLogicalExpire(CACHE_SHOP_KEY,

id,

Result.ok(shop);}这里传入缓存击穿工具类的泛型参数包括

CACHE_SHOP_KEY店铺的关键字id商品的idShop.class店铺的类型-JSON序列化时指定对象的类型this::getById方法的泛型可以使用dbFallback.apply(传入的参数)调用这个也可以定义为任何可执行方法20L逻辑过期时间TimeUnit.SECONDS时间的单位

使用Redis作为分布式锁解决集群下的线程并发安全问题基于消费者组模式完成异步下单功能。

2.1

分布式锁的核心思想就是让大家都使用同一把锁只要大家使用的是同一把锁那么我们就能锁住线程不让线程进行让程序串行执行这就是分布式锁的核心思路

1、解决错误删除的问题2、解决释放锁条件判断成功后出现阻塞把别的线程释放的问题。

基本思路是使用setnx但是为了防止死锁我们加入过期时间。

但这这样如果一个线程阻塞会导致别的线程把自己的锁给释放为此我们加入了线程id的验证。

但是存在一个高并发情况下有可能线程id相同的情况为此我们加入了UUID保证每个线程的唯一。

释放锁的过程必须是原子性因为如果释放过程条件判断是自己线程id的时候后出现阻塞的话也会导致锁的自动释放等别的进程来了之后刚获取的锁会被这个阻塞的释放锁过程释放所以我们采用了lua脚本的方式保证了获取锁和释放锁的原子性。

最后为了进一步提高响应速度我们使用了异步下单的方式提高并发能力。

java代码

利用setnx方法进行加锁同时增加过期时间防止死锁此方法可以保证加锁和增加过期时间具有原子性

private

stringRedisTemplate){this.namename;this.stringRedisTemplate

stringRedisTemplate;

Thread.currentThread().getId()//

获取锁Boolean

stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX

name,

1、由于为了防止死锁问题设计过期时间。

当过期时间后又有线程获取了锁对象之前没有执行完的方法就会释放对象锁。

2、在解决1的情况下如果释放锁的过程中出现了阻塞那么该线程会以为自己一直是要释放自己的锁然而却把别的刚拿到的锁给释放了。

2.2.2

在获取锁时存入线程标示在释放锁时先获取锁中的线程标示判断是否与当前线程标示一致。

如果一致则释放锁如果不一致则不释放锁

Redis提供了Lua脚本功能在一个脚本中编写多条Redis命令确保多条命令执行时的原子性

lua脚本

位置ClassPathResource这个类会默认去classpath下面找我们的lua就放在resources下面滴

“文件名称”UNLOCK_SCRIPT.setLocation(new

ClassPathResource(unlock.lua));UNLOCK_SCRIPT.setResultType(Long.class);//返回值}public

void

调用lua脚本stringRedisTemplate.execute(UNLOCK_SCRIPT,//提前读取文件

会产生IO流

返回Collections.singletonList(KEY_PREFIX

name),ID_PREFIX

Thread.currentThread().getId());//线程表示

以上是我们解决分布式锁的查询的问题该分布式锁具有以下中重点特性

1、利用setnx方法满足互斥性

Executors.newSingleThreadExecutor();//在当前

Impl类初始化之后执行

{SECKILL_ORDER_EXECUTOR.submit(new

当初始化完毕后就会去从队列

2.创建订单handleVoucherOrder(voucherOrder);}

catch

handleVoucherOrder(VoucherOrder

voucherOrder)

redissonClient.getLock(lock:order:

userId);//

获取锁失败直接返回失败或者重试log.error(不允许重复下单);return;}try

{//注意由于是spring的事务是放在threadLocal中此时的是多线程事务会失效proxy.createVoucherOrder(voucherOrder);}

finally

释放锁redisLock.unlock();}}//阻塞队列private

orderTasks

UserHolder.getUser().getId();long

orderId

stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(),

userId.toString(),

redisIdWorker.nextId(order);voucherOrder.setId(orderId);//

2.4.用户idvoucherOrder.setUserId(userId);//

2.5.代金券idvoucherOrder.setVoucherId(voucherId);//

2.6.放入阻塞队列orderTasks.add(voucherOrder);//3.子线程不能获取代理对象主线程可以

(IVoucherOrderService)AopContext.currentProxy();//4.返回订单idreturn

Result.ok(orderId);}Transactional//把他的voucherOrder创建

seckillVoucherpublic

createVoucherOrder(VoucherOrder

voucherOrder)

voucherOrder.getVoucherId()).count();//

(count

用户已经购买过了log.error(用户已经购买过了);return

;}//

seckillVoucherService.update().setSql(stock

stock

voucherOrder.getVoucherId()).gt(stock,

where

1、主线程主要在redis中对使用lua脚本对库存、用户id判断并将创建的订单放到队列中

2.5

先利用Redis完成库存余量、一人一单判断完成抢单业务再将下单业务放入阻塞队列利用独立线程异步下单提高并发量

3、基于Redis中的sortedSet的排序功能实现点赞排行榜用时间戳作为score筛选出最早点赞的用户

Redis的sorted_set是有序集合在set的基础上增加score属性用来排序

点赞功能Overridepublic

UserHolder.getUser().getId();//

key

stringRedisTemplate.opsForZSet().score(key,

(score

{stringRedisTemplate.opsForZSet().add(key,

userId.toString(),

{stringRedisTemplate.opsForZSet().remove(key,

Result.ok();}private

stringRedisTemplate.opsForZSet().score(key,

userId.toString());blog.setIsLike(score

null);}点赞列表查询

stringRedisTemplate.opsForZSet().range(key,

4);if

Result.ok(Collections.emptyList());}//

ids

top5.stream().map(Long::valueOf).collect(Collectors.toList());String

idStr

userService.listByIds(ids).stream().map(user

UserDTO.class)).collect(Collectors.toList());*///

3.根据用户id查询用户

userService.query()//自定义查询.in(id,

ids).last(ORDER

UserDTO.class)).collect(Collectors.toList());//

4.返回return

1、点赞时使用博客的唯一主键作为zset的key使用用户id作为value。

这样每次点赞或者取消点赞可以把相应的用户添加或者移除。

需要注意的是我们在加入点赞信息的时候需要加入时间戳为后面的点赞排序做准备。

zrange

4、使用MongoDB记录用户搜索记录使用ElasticSearch优化搜索功能提高用户体验和减轻数据库压力。

此处是将黑马头条中的自媒体模块的搜索功能加入到了这个外卖app中。

4.1

org.springframework.data.mongodb.core.mapping.Document;import

import

dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-mongodb/artifactId

/dependency②nacos配置

创建Service中新增insert方法并设计实现类。

在实现方法中添加注解Async来实现异步调用

Service

Query.query(Criteria.where(userId).is(userId).and(keyword).is(keyword));ApUserSearch

apUserSearch

null){apUserSearch.setCreatedTime(new

Date());mongoTemplate.save(apUserSearch);return;}//3.不存在判断当前历史记录总数量是否超过10apUserSearch

new

ApUserSearch();apUserSearch.setUserId(userId);apUserSearch.setKeyword(keyword);apUserSearch.setCreatedTime(new

Date());Query

Query.query(Criteria.where(userId).is(userId));query1.with(Sort.by(Sort.Direction.DESC,createdTime));ListApUserSearch

apUserSearchList

ApUserSearch.class);if(apUserSearchList

null

10){mongoTemplate.save(apUserSearch);}else

{ApUserSearch

apUserSearchList.get(apUserSearchList.size()

1);mongoTemplate.findAndReplace(Query.query(Criteria.where(id).is(lastUserSearch.getId())),apUserSearch);}}

}4.1.4

StringUtils.isBlank(dto.getSearchWords())){return

ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);}ApUser

user

AppThreadLocalUtil.getUser();//异步调用

保存搜索记录if(user

0){apUserSearchService.insert(dto.getSearchWords(),

user.getId());}

加载搜索记录列表在mongoDb数据中查询前十条按照时间顺序倒叙展示

实现方法

AppThreadLocalUtil.getUser();if(user

null){return

ResponseResult.errorResult(AppHttpCodeEnum.NEED_LOGIN);}//根据用户查询数据按照时间倒序ListApUserSearch

apUserSearches

mongoTemplate.find(Query.query(Criteria.where(userId).is(user.getId())).with(Sort.by(Sort.Direction.DESC,

createdTime)),

ResponseResult.okResult(apUserSearches);

}4.2

dependencygroupIdorg.elasticsearch.client/groupIdartifactIdelasticsearch-rest-high-level-client/artifactIdversion7.4.0/version

/dependency

dependencygroupIdorg.elasticsearch.client/groupIdartifactIdelasticsearch-rest-client/artifactIdversion7.4.0/version

/dependency

dependencygroupIdorg.elasticsearch/groupIdartifactIdelasticsearch/artifactIdversion7.4.0/version

/dependency2、nacos配置中心添加搜索微服务配置ip和端口

org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration

elasticsearch:host:

StringUtils.isBlank(dto.getSearchWords())){return

ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);}//2.设置查询条件SearchRequest

searchRequest

SearchRequest(app_info_article);SearchSourceBuilder

searchSourceBuilder

SearchSourceBuilder();//布尔查询BoolQueryBuilder

boolQueryBuilder

QueryBuilders.boolQuery();//关键字的分词之后查询QueryStringQueryBuilder

QueryBuilders.queryStringQuery(dto.getSearchWords()).field(title).field(content).defaultOperator(Operator.OR);boolQueryBuilder.must(queryStringQueryBuilder);//查询小于mindate的数据RangeQueryBuilder

rangeQueryBuilder

QueryBuilders.rangeQuery(publishTime).lt(dto.getMinBehotTime().getTime());boolQueryBuilder.filter(rangeQueryBuilder);//分页查询searchSourceBuilder.from(0);searchSourceBuilder.size(dto.getPageSize());//按照发布时间倒序查询searchSourceBuilder.sort(publishTime,

highlightBuilder

HighlightBuilder();highlightBuilder.field(title);highlightBuilder.preTags(font

stylecolor:

inherit;);highlightBuilder.postTags(/font);searchSourceBuilder.highlighter(highlightBuilder);searchSourceBuilder.query(boolQueryBuilder);searchRequest.source(searchSourceBuilder);SearchResponse

searchResponse

restHighLevelClient.search(searchRequest,

RequestOptions.DEFAULT);//3.结果封装返回ListMap

list

searchResponse.getHits().getHits();for

(SearchHit

Map.class);//处理高亮if(hit.getHighlightFields()

null

hit.getHighlightFields().size()

0){Text[]

hit.getHighlightFields().get(title).getFragments();String

title

StringUtils.join(titles);//高亮标题map.put(h_title,title);}else

{//原始标题map.put(h_title,map.get(title));}list.add(map);}return

ResponseResult.okResult(list);}

}解析

1、搜索的参数关键字、时间、第几页ES中使用must和filter连接多个搜索条件

2、所得到的hit文件中会有单独的一个hightlight属性的值这里面设置了高亮的信息我们将其取出后覆盖之前查询的结果就可以实现高亮显示的效果

3、我们在覆盖是一些细节就是getFragments()来获取查询中的所有结果结果是一个分片集合。

使用Kafka完成内部系统的消息通知起到了削峰填谷及解耦的作用。

因为当时我们两个微服务是分开设计的自媒体微服务可以先处理完将信息发送给kafka让app微服务异步去更新信息。

1、我们发送的是文章信息只是topic绑定了app微服务后期我们可以扩展更多的微服务例如大数据之类的

5.1

2、在生产者自媒体端的nacos配置中加入生产者配置。

在app端的nacos中配置消费者配置

生产者

spring:kafka:bootstrap-servers:

192.168.200.130:9092consumer:group-id:

${spring.application.name}key-deserializer:

org.apache.kafka.common.serialization.StringDeserializervalue-deserializer:

org.apache.kafka.common.serialization.StringDeserializer消费者

spring:kafka:bootstrap-servers:

192.168.200.130:9092consumer:group-id:

${spring.application.name}key-deserializer:

org.apache.kafka.common.serialization.StringDeserializervalue-deserializer:

org.apache.kafka.common.serialization.StringDeserializer5.2

Component

apArticleConfigService;KafkaListener(topics

WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC)public

void

message){if(StringUtils.isNotBlank(message)){Map

map

Map.class);apArticleConfigService.updateByMap(map);log.info(article端文章配置修改articleId{},map.get(articleId));}}

基于Feign熔断降级的功能编写降级逻辑防止服务出现雪崩问题。

服务降级是服务自我保护的一种方式或者保护下游服务的一种方式用于确保服务不会受请求突增影响变得不可用确保服务不会崩溃

6.1

feign只是定义了抽象方法方便依赖工程可以使用相应方法以及让别的工程实现它做具体的事情。

类似于动态代理的意思一样把需要做的事情先抽象出方法来。

可以看到我们在定义了feign模块并使用了自媒体模块依赖它在feign定义了保存接口并由文章模块的类实现。

这样的话我们通过自动注入调用feign接口的保存方法相当于实际调用的是文章模块中的保存方法。

1、feign模块定义接口

{PostMapping(/api/v1/article/save)public

ResponseResult

articleClient.saveArticle(dto);3、在需要调用的模块中定一个feign接口的实现类风格为正常controller形式

RestController

apArticleService;PostMapping(/api/v1/article/save)Overridepublic

ResponseResult

apArticleService.saveArticle(dto);}

Overridepublic

ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);}6.2

实现步骤

1、编写降级逻辑的微服务注意这里的降级逻辑是写在feign模块中自我理解

因为刚才远程调用其实也是把接口实现了我们就可以在feign模块中写另外一个实现类在远程调用不成功的时候就可以运行另外一个实现类当然这也是feign给我们提供的功能具体实现是通过

2、自媒体使用feign远程调用

3、在feign远程接口也就是你在feign中抽象出来的方法上面中指向降级代码接口实现类

4、在自媒体微服务中开启服务降级功能

①在heima-leadnews-feign-api编写降级逻辑

package

com.heima.apis.article.fallback;/***

feign失败配置*

ResponseResult.errorResult(AppHttpCodeEnum.SERVER_ERROR,获取数据失败);}

package

com.heima.wemedia.config;import

org.springframework.context.annotation.ComponentScan;

import

org.springframework.context.annotation.Configuration;Configuration

ComponentScan(com.heima.apis.article.fallback)

public

com.heima.apis.article.fallback.IArticleClientFallback;

import

com.heima.model.article.dtos.ArticleDto;

import

com.heima.model.common.dtos.ResponseResult;

import

org.springframework.cloud.openfeign.FeignClient;

import

org.springframework.web.bind.annotation.PostMapping;

import

org.springframework.web.bind.annotation.RequestBody;FeignClient(value

public

{PostMapping(/api/v1/article/save)public

ResponseResult

}③客户端开启降级heima-leadnews-wemedia

在wemedia的nacos配置中心里添加如下内容开启服务降级也可以指定服务响应的超时的时间

feign:#

开启feign对hystrix熔断降级的支持hystrix:enabled:

true#

修改调用超时时间client:config:default:connectTimeout:

2000readTimeout: