课程信息编辑完毕即可发布课程,发布课程相当于—个确认操作,课程发布后学习者在网站可以搜索到课程,然后查看课程的详细信息,进一步选课、支付、在线学习。

下边是课程编辑与发布的整体流程:

为了课程内容没有违规信息、课程内容安排合理,在课程发布之前运营方会进行课程审核,审核通过后课程方可发布。

作为课程制作方即教学机构,在课程发布前通过课程预览功能可以看到课程发布后的效果,哪里的课程信息存在问题方便查看,及时修改。

Freemarker

课程预览页面使用Freemarker技术,后端填充数据返回给前端

在Nacos中添加Freemarker配置,创建内容管理工程配置:content-api-dev.yaml

  • ID:freemarker-config-dev.yaml

  • Group:learning-online-common

  • 描述:静态模版引擎Freemarker

  • 配置内容:

    freemarker-config-dev.yaml
    spring:
    freemarker:
    enabled: true
    cache: false # 关闭模版缓存,方便测试
    settings:
    template_update_delay: 0
    suffix: .ftl
    charset: UTF-8
    template-loader-path: classpath:/templates/
    resources:
    add-mappings: false # 关闭项目中的静态资源映射

在内容管理工程的learning-online-content-api模块的pom文件中添加依赖:

pom.xml
<!-- 静态模版引擎Freemarker -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-freemarker</artifactId>
</dependency>

修改learning-online-content-api模块的配置文件,引入freemarker配置

bootstrap.yml
shared-configs:
- data-id: freemarker-config-${spring.profiles.active}.yaml
group: learning-online-common
refresh: true

课程预览服务

resources目录下创建templates目录,并在目录下放入course_template.ftl模版文件,该文件下载地址:

接口信息如下

路径地址 http://localhost:63040/content/coursepreview/267
请求方式 GET
请求参数 Long
返回结果 ModelAndView

定义CoursePreviewVO

CoursePreviewVO
/**
* 课程预览
*/
@Data
public class CoursePreviewVO {

// 课程基本信息, 营销信息
private CourseBaseInfoVO courseBase;

// 课程计划信息
private List<TeachPlanVO> teachplans;
}

定义Service

找到CoursePublishService接口,定义计划查询接口

CoursePublishService
/**
* <p>
* 课程发布 服务类
* </p>
*
* @author sw-code
* @since 2023-08-18
*/
public interface CoursePublishService extends IService<CoursePublish> {

/**
* 获取课程预览信息
*
* @param courseId 课程ID
* @return CoursePreviewVO 预览信息
*/
public CoursePreviewVO getCoursePreviewInfo(Long courseId);

}

实现该方法,找到其实现类CoursePublishServiceImpl

CoursePublishServiceImpl
/**
* <p>
* 课程发布 服务实现类
* </p>
*
* @author sw-code
* @since 2023-08-18
*/
@Slf4j
@Service
public class CoursePublishServiceImpl extends ServiceImpl<CoursePublishMapper, CoursePublish> implements CoursePublishService {

private final CourseBaseService courseBaseService;
private final TeachPlanService teachPlanService;

public CoursePublishServiceImpl(CourseBaseService courseBaseService, TeachPlanService teachPlanService) {
this.courseBaseService = courseBaseService;
this.teachPlanService = teachPlanService;
}

/**
* 获取课程预览信息
*
* @param courseId 课程ID
* @return CoursePreviewVO 预览信息
*/
@Override
public CoursePreviewVO getCoursePreviewInfo(Long courseId) {
// 课程基本信息, 营销信息
CourseBaseInfoVO courseBaseInfo = courseBaseService.getCourseBaseInfo(courseId);
// 课程计划信息
List<TeachPlanVO> teachPlans = teachPlanService.getTreeNodes(courseId);
// 封装VO
CoursePreviewVO previewVO = new CoursePreviewVO();
previewVO.setCourseBase(courseBaseInfo);
previewVO.setTeachplans(teachPlans);
return previewVO;
}
}

定义Controller

注意返回的是ModelView模版数据,不能使用@RestController

TeachPlanController
@Controller
public class CoursePublishController {

private final CoursePublishService coursePublishService;

public CoursePublishController(CoursePublishService coursePublishService) {
this.coursePublishService = coursePublishService;
}

@GetMapping("/coursepreview/{courseId}")
public ModelAndView preview(@PathVariable("courseId") @NotNull(message = "课程ID不能为空") Long courseId) {
ModelAndView data = new ModelAndView();
CoursePreviewVO previewInfo = coursePublishService.getCoursePreviewInfo(courseId);
data.addObject("model", previewInfo);
data.setViewName("course_template");
return data;
}
}

课程预览服务以提供,打开前端项目查看是否成功

Push到Git

commit "完成课程预览功能"

视频预览服务

接口信息如下

查询课程信息,无需登陆操作

路径地址 http://localhost:63040/content/open/course/whole/2
请求方式 GET
请求参数 Long
返回结果 CoursePreviewVO

查询视频URL

路径地址 http://localhost:63040/content/open/preview/mediaId
请求方式 GET
请求参数 String
返回结果 R

定义Controller

找到learning-online-content-api模块,创建如下Controller

CourseOpenController
@RestController
@RequestMapping("/open")
public class CourseOpenController {

private final CoursePublishService coursePublishService;

public CourseOpenController(CoursePublishService coursePublishService) {
this.coursePublishService = coursePublishService;
}

@GetMapping("/course/whole/{courseId}")
public CoursePreviewVO preview(@PathVariable("courseId") @NotNull(message = "课程ID不能为空") Long courseId) {
return coursePublishService.getCoursePreviewInfo(courseId);
}

}

找到learning-online-media-api模块,创建如下Controller

MediaOpenController
/**
* 视频播放页面,openapi
*/

@RestController
@ResponseResult
@RequestMapping("/open")
public class MediaOpenController {

private final MediaFilesService mediaFilesService;

public MediaOpenController(MediaFilesService mediaFilesService) {
this.mediaFilesService = mediaFilesService;
}

@GetMapping("/preview/{mediaId}")
public R getPlayUrlByMediaId(@PathVariable("mediaId") String mediaId) {
MediaFiles mediaFiles = mediaFilesService.getById(mediaId);
if (mediaFiles == null || StringUtils.isEmpty(mediaFiles.getUrl())) {
throw new BizException("视频还没有转码处理");
}
return R.success(mediaFiles.getUrl());
}
}

课程预览服务以提供,打开前端项目查看是否成功

Push到Git

commit "完成视频预览功能"

课程审核服务

在课程基本表course_base表设置课程审核状态字段,包括:未提交、已提交(未审核)、审核通过、审核不通过。

下边是课程状态的转化关系:

说明如下:

1、一门课程新增后它的审核状为”未提交“,发布状态为”未发布“。

2、课程信息编辑完成,教学机构人员执行”提交审核“操作。此时课程的审核状态为”已提交“。

3、当课程状态为已提交时运营平台人员对课程进行审核。

4、运营平台人员审核课程,结果有两个:审核通过、审核不通过。

5、课程审核过后不管状态是通过还是不通过,教学机构可以再次修改课程并提交审核,此时课程状态为”已提交“。此时运营平台人员再次审核课程。

6、课程审核通过,教学机构人员可以发布课程,发布成功后课程的发布状态为”已发布“。

7、课程发布后通过”下架“操作可以更改课程发布状态为”下架“

8、课程下架后通过”上架“操作可以再次发布课程,上架后课程发布状态为“发布”。

审核过程中,用户的修改操作是被允许的,如果审核和用户共同查询课程信息表,会造成冲突。为了避免冲突,在用户点击审核时,将课程信息拷贝到预审表。审核通过后修改预发布表和基本信息表的审核状态为审核通过,发布时课程信息数据则从预发布表中获取, 流程如下:

关于审核部分本项目不予实现,这里只实现将审核信息放入预审表

定义Service

找到CoursePublishService接口,定义提交审核接口

CoursePublishService
/**
* 提交审核
* 将课程基本信息、课程计划信息和营销信息保存到预发布表
*
* @param companyId 机构ID
* @param courseId 课程ID
*/
public void commitAudit(Long companyId, Long courseId);

实现该方法,找到其实现类CoursePublishServiceImpl

CoursePublishServiceImpl
/**
* 提交审核
* 将课程基本信息、课程计划信息和营销信息保存到预发布表
*
* @param companyId 机构ID
* @param courseId 课程ID
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void commitAudit(Long companyId, Long courseId) {
// 查询课程基本信息
CourseBase courseBase = Optional.ofNullable(courseBaseService.getById(courseId)).orElseThrow(() -> new BizException("课程信息不存在"));
// 如果课程的审核状态为已提交则不允许提交
if ("202003".equals(courseBase.getAuditStatus())) {
throw new BizException("当前为等待审核状态,审核完成可以再次提交");
}
if (!courseBase.getCompanyId().equals(companyId)) {
throw new BizException("不允许提交其它机构的课程");
}
// 课程的图片
if (StringUtils.isEmpty(courseBase.getPic())) {
throw new BizException("提交失败,请上传课程图片");
}

// 查询课程计划信息
List<TeachPlanVO> teachPlans = teachPlanService.getTreeNodes(courseId);
// 计划信息没有填写也不允许提交
if (teachPlans.isEmpty()) {
throw new BizException("提交失败,还没有添加课程计划");
}

// 查询营销信息
CourseMarket courseMarket = Optional.ofNullable(courseMarketService.getById(courseId)).orElseThrow(() -> new BizException("请完善课程营销信息"));

// 查询分类信息
List<String> list = Arrays.asList(courseBase.getSt(), courseBase.getMt());
String join = StringUtils.join(list, ",");
List<CourseCategory> categories = courseCategoryService.list(Wrappers.<CourseCategory>lambdaQuery()
.in(CourseCategory::getId, list).last("ORDER BY FIELD(id, '" + list.get(0) + "', '" + list.get(1) + "')"));

// 查询课程基本信息、营销信息、计划等信息插入到课程预发布表
CoursePublishPre coursePublishPre = new CoursePublishPre();
// 设置分类名
coursePublishPre.setStName(categories.get(0) == null ? "" : categories.get(0).getName());
coursePublishPre.setMtName(categories.get(1) == null ? "" : categories.get(1).getName());
BeanUtils.copyProperties(courseMarket, coursePublishPre);
// 设置基本信息
BeanUtils.copyProperties(courseBase, coursePublishPre);
// 设置营销信息
coursePublishPre.setMarket(JSON.toJSONString(courseMarket));
// 设置课程计划
coursePublishPre.setTeachplan(JSON.toJSONString(teachPlans));

//设置预发布记录状态,已提交
coursePublishPre.setStatus("202003");
//教学机构id
coursePublishPre.setCompanyId(companyId);
//提交时间
coursePublishPre.setCreateDate(LocalDateTime.now());

// 保存预发布信息
CoursePublishPre coursePublishPreUpdate = coursePublishPreService.getById(courseId);
if (coursePublishPreUpdate == null) {
//添加课程预发布记录
coursePublishPreService.save(coursePublishPre);
} else {
coursePublishPreService.updateById(coursePublishPre);
}

//更新课程基本表的审核状态
courseBase.setAuditStatus("202003");
courseBaseService.updateById(courseBase);

}

定义Controller

注意返回的是ModelView模版数据,不能使用@RestController

CoursePublishController
@Api(value = "课程发布审核接口", tags = "课程发布审核接口")
@ResponseResult
@RestController
public class CoursePublishController {

private final CoursePublishService coursePublishService;

public CoursePublishController(CoursePublishService coursePublishService) {
this.coursePublishService = coursePublishService;
}

@ApiOperation("提交审核")
@PostMapping("/courseaudit/commit/{courseId}")
public void commitAudit(@PathVariable("courseId") Long courseId) {
Long companyId = 1232141425L;
coursePublishService.commitAudit(companyId, courseId);
}
}

课程发布服务

该部分需要实现媒资管理服务的文件上传,请先转到媒资管理工程,完成该部分开发。

课程发布时,会将数据缓存至Redis、将添加Elasticsearch索引信息(便于检索)、生成静态化文件并存储到MinIO中。因此为了保证事务一致,项目采用AP模式,即保证高可用。AP模式的实现方式为,使用任务调度的方案,启动任务调度将课程信息由数据库同步到elasticsearch、MinlO、redis中。重试机制保证消息最终一致。

分布式事务

分布式事务控制有哪些常用的技术方案?

实现CP就是要实现强一致性:

  • 使用Seata框架基于AT模式实现
  • 使用Seata框架基于TCC模式实现

实现AP则要保证最终数据一致性:

  • 使用消息队列通知的方式去实现,通知失败自动重试,达到最大失败次数需要人工处理;
  • 使用任务调度的方案,启动任务调度将课程信息由数据库同步到elasticsearch、MinlO、redis中。

本项目使用AP方式,最终的流程如下:

1、在内容管理服务的数据库中添加一个消息表,消息表和课程发布表在同一个数据库。

2、点击课程发布通过本地事务向课程发布表写入课程发布信息,同时向消息表写课程发布的消息。通过数据库进行控制,只要课程发布表插入成功消息表也插入成功,消息表的数据就记录了某门课程发布的任务。

3、启动任务调度系统定时调度内容管理服务去定时扫描消息表的记录。

4、当扫描到课程发布的消息时即开始完成向redis、elasticsearch、MinIO同步数据的操作。

5、同步数据的任务完成后删除消息表记录。

时序图如下:

发布课程

用户点击课程发布时,需要将预发布表中的数据转移到发布表中,同时向消息表中添加记录

为更方便使用任务调度,引入learning-online-message-sdk,封装好的调度SDK。其会对lo_content数据库的mq_messagemq_message_history两张消息记录表进行增删改查操作。

定义Service

定义接口方法

CoursePublishService
/**
* 发布课程
*
* @param companyId 机构ID
* @param courseId 课程ID
*/
public void publish(Long companyId, Long courseId);

使用该方法:

CoursePublishServiceImpl
private final CoursePublishPreService coursePublishPreService;
private final CourseBaseService courseBaseService;
private final MqMessageService mqMessageService;

/**
* 发布课程
*
* @param companyId 机构ID
* @param courseId 课程ID
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void publish(Long companyId, Long courseId) {
// 查询预发布表
CoursePublishPre coursePublishPre = Optional.ofNullable(coursePublishPreService.getById(courseId))
.orElseThrow(() -> new BizException("未提交审核,不允许发布"));

if (!coursePublishPre.getCompanyId().equals(companyId)) {
throw new BizException("不允许发布其它机构的课程");
}
if (!coursePublishPre.getStatus().equals("202004")) {
// 未通过审核
throw new BizException("未通过审核,不允许发布");
}
// 向课程发布表写数据
CoursePublish coursePublish = new CoursePublish();
BeanUtils.copyProperties(coursePublishPre, coursePublish);
// 查询是否有发布记录
CoursePublish dbCoursePublish = getById(courseId);
if (dbCoursePublish == null) {
// 没有则新增
save(coursePublish);
} else {
// 有则更新
updateById(coursePublish);
}

// 更改文章状态为已经发布
courseBaseService.update(Wrappers.<CourseBase>lambdaUpdate().eq(CourseBase::getId, courseId).set(CourseBase::getStatus, "203002"));

// 向消息表写入数据
saveCoursePublishMessage(courseId);

// 删除预发布表数据
coursePublishPreService.removeById(coursePublishPre);
}

/**
* 保存消息记录
*
* @param courseId 课程id
*/
public void saveCoursePublishMessage(Long courseId) {
// 使用消息SDK插入课程发布消息
Optional.ofNullable(mqMessageService.addMessage("course_publish", String.valueOf(courseId), null, null))
.orElseThrow(() -> new BizException("保存消息记录失败"));
}

定义Controller

CoursePublishController
@ApiOperation("课程发布")
@PostMapping("/coursepublish/{courseId}")
public void coursePublish(@PathVariable("courseId") Long courseId) {
Long companyId = 1232141425L;
coursePublishService.publish(companyId, courseId);
}

任务调度

引入依赖

为了使用任务调度,需要在learning-online-content-service模块中添加如下依赖:

pom.xml
<dependency>
<groupId>com.xuxueli</groupId>
<artifactId>xxl-job-core</artifactId>
</dependency>

配置信息

按照XXL-JOB的使用方式,还需要进行配置:

首先添加Nacos的配置:

content-service-dev.yaml
# xxl-job配置
xxl:
job:
admin:
# 调度中心部署跟地址 [选填]:如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调";为空则关闭自动注册;
addresses: http://127.0.0.1:8080/xxl-job-admin
executor:
# 执行器注册 [选填]:优先使用该配置作为注册地址,为空时使用内嵌服务 ”IP:PORT“ 作为注册地址。从而更灵活的支持容器类型执行器动态IP和动态映射端口问题。
address:
# 执行器AppName [选填]:执行器心跳注册分组依据;为空则关闭自动注册
appname: coursepublish-executor
# 执行器IP [选填]:默认为空表示自动获取IP,多网卡时可手动设置指定IP,该IP不会绑定Host仅作为通讯实用;地址信息用于 "执行器注册" 和 "调度中心请求并触发任务";
ip:
# 执行器端口号 [选填]:小于等于0则自动获取;默认端口为9999,单机部署多个执行器时,注意要配置不同执行器端口;
port: 8999
# 执行器运行日志文件存储磁盘路径 [选填] :需要对该路径拥有读写权限;为空则使用默认路径;
logpath: /Users/swcode/data/applogs/xxl-job/jobhandler
# 执行器日志文件保存天数 [选填] : 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能;
logretentiondays: 30
# 执行器通讯TOKEN [选填]:非空时启用;
accessToken:

在项目中创建配置类:

XxlJonConfig
@Slf4j
@Configuration
public class XxlJonConfig {

@Value("${xxl.job.admin.addresses}")
private String adminAddresses;

@Value("${xxl.job.executor.appname}")
private String appname;

@Value("${xxl.job.executor.address}")
private String address;

@Value("${xxl.job.executor.ip}")
private String ip;

@Value("${xxl.job.executor.port}")
private int port;

@Value("${xxl.job.accessToken}")
private String accessToken;

@Value("${xxl.job.executor.logpath}")
private String logPath;

@Value("${xxl.job.executor.logretentiondays}")
private int logRetentionDays;


@Bean
public XxlJobSpringExecutor xxlJobExecutor() {
log.info(">>>>>>>>>>> xxl-job config init.");
XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
xxlJobSpringExecutor.setAppname(appname);
xxlJobSpringExecutor.setAddress(address);
xxlJobSpringExecutor.setIp(ip);
xxlJobSpringExecutor.setPort(port);
xxlJobSpringExecutor.setAccessToken(accessToken);
xxlJobSpringExecutor.setLogPath(logPath);
xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
return xxlJobSpringExecutor;
}
}

创建执行器:

创建调度任务:

任务处理类

下一步就是定义任务的处理类,这里我们使用了SDK,继承MessageProcessAbstract抽象类,该抽象类实现了XXL-JOB分片广播调度的分片操作、任务拉取、根据系统核心数开启线程(多线程)、线程计数器等基础操作;我们需要重写其中的execute方法,实现具体的业务处理逻辑注意幂等

创建包com.swx.content.service.jobhandler,在该包下创建如下类:

@Slf4j
@Service
public class CoursePublishTask extends MessageProcessAbstract {

private final CoursePublishService coursePublishService;

public CoursePublishTask(CoursePublishService coursePublishService) {
this.coursePublishService = coursePublishService;
}

@XxlJob("CoursePublishJobHandler")
public void coursePublishJobHandler() {
// 分片参数
int shardIndex = XxlJobHelper.getShardIndex();
int shardTotal = XxlJobHelper.getShardTotal();

// 执行任务
this.process(shardIndex, shardTotal, "course_publish", 30, 60);

}

/**
* 任务处理
*
* @param mqMessage 执行任务内容
* @return boolean true:处理成功,false处理失败
*/
@Override
public boolean execute(MqMessage mqMessage) {
if (StringUtils.isEmpty(mqMessage.getBusinessKey1())) {
log.debug("消息参数错误,无业务ID: {}", mqMessage);
return false;
}
long courseId = Long.parseLong(mqMessage.getBusinessKey1());

// 生成并上传静态页面
generateCourseHtml(mqMessage, courseId);
// 创建索引
saveCourseIndex(mqMessage, courseId);
// 缓存至Redis
save2Redis(mqMessage, courseId);
return true;
}

/**
* 生成静态化页面上传至文件系统
*
* @param mqMessage 消息
* @param courseId 课程ID
*/
private void generateCourseHtml(MqMessage mqMessage, long courseId) {
Long taskId = mqMessage.getId();
MqMessageService mqMessageService = this.getMqMessageService();
// 任务幂等性处理
int stageOne = mqMessageService.getStageOne(taskId);
if (stageOne > 0) {
log.debug("静态化页面已生成,无需处理");
return;
}
// 生成静态页面
// 上传静态页面
// 置状态为已完成
mqMessageService.completedStageOne(taskId);
}

/**
* 保存课程索引到Elasticsearch
*
* @param mqMessage 消息
* @param courseId 课程ID
*/
private void saveCourseIndex(MqMessage mqMessage, long courseId) {
Long taskId = mqMessage.getId();
MqMessageService mqMessageService = this.getMqMessageService();
// 任务幂等性处理
int stageTwo = mqMessageService.getStageTwo(taskId);
if (stageTwo > 0) {
log.debug("课程索引已缓存,无需处理");
return;
}
// 创建索引
// 置状态为已完成
mqMessageService.completedStageTwo(taskId);
}

/**
* 缓存到Redis
*
* @param mqMessage 消息
* @param courseId 课程ID
*/
private void save2Redis(MqMessage mqMessage, long courseId) {
Long taskId = mqMessage.getId();
MqMessageService mqMessageService = this.getMqMessageService();
// 任务幂等性处理
int stageThree = mqMessageService.getStageThree(taskId);
if (stageThree > 0) {
log.debug("课程已缓存,无需处理");
return;
}
// 保存到Redis
// 置状态为已完成
mqMessageService.completedStageThree(taskId);
}
}

静态化页面

使用Freemarker技术生成静态化页面,并使用Feign远程调用媒资上传服务,上传至MinIO中。

配置Feign

为了使用Feign调用媒资服务上传文件,需要在learning-online-content-service引入如下依赖:

pom.xml
<!-- Feign远程调用 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
<!-- Feign熔断降级 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
</dependency>
<dependency>
<groupId>io.github.openfeign</groupId>
<artifactId>feign-httpclient</artifactId>
</dependency>
<!-- Feign支持文件传输 -->
<dependency>
<groupId>io.github.openfeign.form</groupId>
<artifactId>feign-form</artifactId>
<version>3.8.0</version>
</dependency>
<dependency>
<groupId>io.github.openfeign.form</groupId>
<artifactId>feign-form-spring</artifactId>
</dependency>

config包下添加文件传输支持配置类:

MultipartSupportConfig
/**
* 传输文件支持
*/
@Configuration
public class MultipartSupportConfig {

private final ObjectFactory<HttpMessageConverters> messageConverters;

public MultipartSupportConfig(ObjectFactory<HttpMessageConverters> messageConverters) {
this.messageConverters = messageConverters;
}

@Bean
@Primary//注入相同类型的bean时优先使用
@Scope("prototype")
public Encoder feignEncoder() {
return new SpringFormEncoder(new SpringEncoder(messageConverters));
}

/**
* 将file转为Multipart
*
* @param file 文件
* @return MultipartFile
*/
public static MultipartFile getMultipartFile(File file) {
FileItem item = new DiskFileItemFactory().createItem("file", MediaType.MULTIPART_FORM_DATA_VALUE, true, file.getName());
try (FileInputStream inputStream = new FileInputStream(file);
OutputStream outputStream = item.getOutputStream();) {
IOUtils.copy(inputStream, outputStream);

} catch (Exception e) {
e.printStackTrace();
}
return new CommonsMultipartFile(item);
}

/**
* 将 输入流 转为Multipart
*
* @param inputStream 输入流
* @param filename 文件名
* @return MultipartFile
*/
public static MultipartFile getMultipartFile(InputStream inputStream, String filename) {
FileItem item = new DiskFileItemFactory().createItem("file", MediaType.MULTIPART_FORM_DATA_VALUE, true, filename);
try (OutputStream outputStream = item.getOutputStream()) {
IOUtils.copy(inputStream, outputStream);
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
inputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
return new CommonsMultipartFile(item);
}
}

在Nacos中新建Feign的配置文件

创建内容管理工程配置:feign-dev.yaml

  • ID:feign-dev.yaml

  • Group:learning-online-common

  • 配置内容:

    feign-dev.yaml
    feign:
    httpclient:
    enabled: true
    hystrix:
    enabled: true
    circuitbreaker:
    enabled: true

    hystrix:
    command:
    default:
    execution:
    isolation:
    #strategy: SEMAPHORE
    thread:
    timeoutInMilliseconds: 1000

    ribbon:
    ConnectTimeout: 60000 # 连接超时时间
    ReadTimeout: 60000
    MaxAutoRetries: 0
    MaxAutoRetriesNextServer: 1

content-apibootstrap.yaml中添加Feign的配置:

- data-id: feign-${spring.profiles.active}.yaml
group: learning-online-common
refresh: true

修改媒资服务

修改媒资服务的文件上传服务,增加一个objectName参数

MediaFilesUploadController
@ApiOperation("上传文件")
@PostMapping(value = "/coursefile", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public UploadFileResultVO upload(@RequestPart("filedata") MultipartFile filedata,
@RequestParam(value = "objectName", required = false) String objectName) {
Long companyId = 1232141425L;
UploadFileParamDTO dto = new UploadFileParamDTO();
dto.setFilename(filedata.getOriginalFilename());
dto.setFileSize(filedata.getSize());
return mediaFilesService.uploadFile(companyId, dto, filedata, objectName);
}
MediaFilesService
/**
* 上传文件
*
* @param companyId 机构ID
* @param dto 文件信息
* @param multipartFile 文件信息
* @param objectName 文件路径
* @return UploadFileResultVO
*/
public UploadFileResultVO uploadFile(Long companyId, UploadFileParamDTO dto, MultipartFile multipartFile, String objectName);
MediaFilesServiceImpl
/**
* 上传文件
*
* @param companyId 机构ID
* @param dto 文件信息
* @param multipartFile 文件信息
* @param objectName 文件路径
* @return UploadFileResultVO
*/
@Override
public UploadFileResultVO uploadFile(Long companyId, UploadFileParamDTO dto, MultipartFile multipartFile, String objectName) {
....;
try {
...;
// 已存在数据库,直接返回
....;
// 获取文章类型
String mineType = getMineType(suffix);
if (mineType.contains("image")) {
dto.setFileType("001001");
} else {
dto.setFileType("001003");
}

// 上传到MinIO
Map<String, String> result = null;
if (StringUtils.hasText(objectName)) {
result = fileStorageService.uploadMediaFile(objectName, mineType, multipartFile.getInputStream());
} else {
result = fileStorageService.uploadMediaFile("", prefix + suffix, mineType, multipartFile.getInputStream());
}

....;
// 保存到数据库,使用编程式事务
....;
// 返回数据
....;
} catch (IOException e) {
....;
}
}

修改MinIO上传方法的逻辑

MinIOFileStorageService
/**
* @param dirPath
* @param filename yyyy/mm/dd/file.jpg
* @return
*/
public String builderFilePath(String dirPath, String filename) {
StringBuilder stringBuilder = new StringBuilder(50);
if (!StringUtils.isEmpty(dirPath)) {
stringBuilder.append(dirPath).append(separator);
}
SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd");
String todayStr = sdf.format(new Date());
stringBuilder.append(todayStr).append(separator);
stringBuilder.append(filename);
return stringBuilder.toString();
}

/**
* 上传文件
*
* @param prefix 文件前缀
* @param filename 文件名
* @param inputStream 文件流
* @return 文件全路径
*/
@Override
public Map<String, String> uploadMediaFile(String prefix, String filename, String mineType, InputStream inputStream) {
String filepath = builderFilePath(prefix, filename);
String bucket = upload2Files(filepath, mineType, inputStream);
Map<String, String> map = new HashMap<>();
map.put("path", filepath);
map.put("bucket", bucket);
return map;
}

/**
* 上传文件
*
* @param objectName 文件路径
* @param inputStream 文件流
* @return 桶和文件全路径
*/
@Override
public Map<String, String> uploadMediaFile(String objectName, String mineType, InputStream inputStream) {
String bucket = upload2Files(objectName, mineType, inputStream);
Map<String, String> map = new HashMap<>();
map.put("path", objectName);
map.put("bucket", bucket);
return map;
}

媒资Client

learning-online-content-service下创建包com.swx.content.client,并创建Feign的接口

/**
* 远程调用媒资服务
*/
@FeignClient(value = "media-api", configuration = MultipartSupportConfig.class, fallbackFactory = MediaServiceClientFallbackFactory.class)
public interface MediaServiceClient {

/**
* 文件传输
*
* @param filedata 文件
* @param objectName MinIO路径
* @return String
*/
@PostMapping(value = "/media/upload/coursefile", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public String upload(@RequestPart("filedata") MultipartFile filedata,
@RequestParam(value = "objectName", required = false) String objectName);
}

同一包下创建熔断降级处理类:

MediaServiceClientFallbackFactory
// 注意别导错包
import org.springframework.cloud.openfeign.FallbackFactory;

/**
* 降级方法,可以拿到异常信息
*/
@Slf4j
@Component
public class MediaServiceClientFallbackFactory implements FallbackFactory<MediaServiceClient> {

@Override
public MediaServiceClient create(Throwable throwable) {
return new MediaServiceClient() {
@Override
public String upload(MultipartFile filedata, String objectName) {
log.debug("远程调用媒资上传文件的接口发生熔断: ", throwable);
return null;
}
};
}
}

定义Service

找到CoursePublishService,分别定义生成和上传课程静态化页面的方法。

/**
* 生成课程静态化页面
*
* @param courseId 课程ID
* @return InputStream
*/
public InputStream generateCourseHtml(Long courseId);

/**
* 上传课程静态化页面
*
* @param courseId 课程ID
* @param inputStream 文件流
*/
public void uploadCourseHtml(Long courseId, InputStream inputStream);

/**
* 查询已发布课程的信息
*
* @param courseId 课程ID
* @return com.swx.content.model.po.CoursePublish 课程发布信息
*/
public CoursePreviewVO getCoursePublish(Long courseId);

CoursePublishServiceImpl中实现方法:

CoursePublishServiceImpl
private final static String COURSE_TEMPLATE = "course_template";
private final static String HTML_PATH = "course/";

/**
* 生成课程静态化页面
*
* @param courseId 课程ID
* @return File
*/
@Override
public InputStream generateCourseHtml(Long courseId) {
try {
Template template = configuration.getTemplate(COURSE_TEMPLATE + ".ftl");
StringWriter out = new StringWriter();
// 数据模型
HashMap<String, Object> params = new HashMap<>();
// 获取发布课程的信息,作为静态页面的参数
params.put("model", getCoursePublish(courseId));
// 文章内容通过freemarker生成html文件
template.process(params, out);
// 返回文件流
return new ByteArrayInputStream(out.toString().getBytes());
} catch (Exception e) {
log.error("页面静态化出现问题", e);
}
return null;
}

/**
* 上传课程静态化页面
*
* @param courseId 课程ID
* @param inputStream 文件流
*/
@Override
public void uploadCourseHtml(Long courseId, InputStream inputStream) {
// 把html文件上传到minio中
MultipartFile multipartFile = MultipartSupportConfig.getMultipartFile(inputStream, courseId + COURSE_TEMPLATE + ".html");
// Feign调用接口
String upload = mediaServiceClient.upload(multipartFile, HTML_PATH + courseId + ".html");
if (upload == null) {
log.debug("上传结果为null, 课程ID: {}", courseId);
}
}

/**
* 查询已发布课程的信息
*
* @param courseId 课程ID
* @return com.swx.content.model.po.CoursePublish 课程发布信息
*/
public CoursePreviewVO getCoursePublish(Long courseId) {
CoursePublish coursePublish = getById(courseId);
if (coursePublish == null) {
throw new BizException("课程未发布");
}
CoursePreviewVO coursePreviewVO = new CoursePreviewVO();

// 解析课程基本信息和营销信息
CourseBaseInfoVO courseBaseInfoVO = new CourseBaseInfoVO();
BeanUtils.copyProperties(coursePublish, courseBaseInfoVO);
coursePreviewVO.setCourseBase(courseBaseInfoVO);

// 解析课程计划信息
List<TeachPlanVO> teachPlans = JSON.parseArray(coursePublish.getTeachplan(), TeachPlanVO.class);
coursePreviewVO.setTeachplans(teachPlans);

return coursePreviewVO;
}

完善调度任务逻辑

找到文章发布任务调度类CoursePublishTask,修改生成静态化页面上传至文件系统逻辑

CoursePublishTask
/**
* 生成静态化页面上传至文件系统
*
* @param mqMessage 消息
* @param courseId 课程ID
*/
private void generateCourseHtml(MqMessage mqMessage, long courseId) {
Long taskId = mqMessage.getId();
MqMessageService mqMessageService = this.getMqMessageService();
// 任务幂等性处理
int stageOne = mqMessageService.getStageOne(taskId);
if (stageOne > 0) {
log.debug("静态化页面已生成,无需处理");
return;
}
// 生成静态页面
InputStream inputStream = coursePublishService.generateCourseHtml(courseId);
if (inputStream == null) {
return;
}
// 上传静态页面
coursePublishService.uploadCourseHtml(courseId, inputStream);
// 置状态为已完成
mqMessageService.completedStageOne(taskId);

}

修改Nginx配置

新增如下配置信息,将其代理到文件服务的course目录下

learning.conf
location /course/ {
proxy_pass http://fileserver/mediafiles/course/;
}

测试

生成的静态文件将存储在MinIO中的mediafiles/course/courseId.html

浏览器访问:http://www.51xuecheng.cn/course/2.html

创建内容索引

ElasticSearch

MySQL Elasticsearch 说明
Table Index 索引(index),就是文档的集合,类似数据库的表(table)
Row Document 文档(Document),就是一条条的数据,类似数据库中的行(Row),文档都是JSON格式
Column Field 字段(Field),就是JSON文档中的字段,类似数据库中的列(Column)
Schema Mapping Mapping(映射)是索引中文档的约束,例如字段类型约束。类似数据库的表结构(Schema)
SQL DSL DSL是elasticsearch提供的JSON风格的请求语句,用来操作elasticsearch,实现CRUD

拉取ElasticSearch镜像

docker pull elasticsearch:7.4.0

创建ElasticSearch容器

docker run -id --name elasticsearch \
-p 9200:9200 -p 9300:9300 \
-v /usr/share/elasticsearch/plugins:/usr/share/elasticsearch/plugins \
-e "discovery.type=single-node" elasticsearch:7.4.0

配置中文分词器 ik

下载地址:https://github.com/medcl/elasticsearch-analysis-ik/releases

上传到服务器的/usr/share/elasticsearch/plugins,解压到新建目录analysis-ik

创建索引

mapping是对索引库中文档的约束,常见的mapping属性包括:

  • type:字段数据类型,常见的简单类型有:

    • 字符串:text(可分词的文本)、keyword(精确值,例如:品牌、国家等)

    • 数值:long、integer、short、byte、double、float、

    • 布尔:boolean

    • 日期:date

    • 对象:object

  • index:是否创建索引,默认为true

  • analyzer:使用哪种分词器

  • properties:该字段的子字段

创建索引course_publish,如下:

{
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0
},
"mappings": {
"properties": {
"id": {
"type": "keyword"
},
"companyId": {
"type": "keyword"
},
"companyName": {
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_smart"
},
"name": {
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_smart"
},
"users": {
"type": "text",
"index": false
},
"tags": {
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_smart"
},
"mt": {
"type": "keyword"
},
"mtName": {
"type": "keyword"
},
"st": {
"type": "keyword"
},
"stName": {
"type": "keyword"
},
"grade": {
"type": "keyword"
},
"teachmode": {
"type": "keyword"
},
"pic": {
"type": "text",
"index": false
},
"description": {
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_smart"
},
"createDate": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss"
},
"status": {
"type": "keyword"
},
"remark": {
"type": "text",
"index": false
},
"charge": {
"type": "keyword"
},
"price": {
"type": "scaled_float",
"scaling_factor": 100.0
},
"originalPrice": {
"type": "scaled_float",
"scaling_factor": 100.0
},
"validDays": {
"type": "integer"
}
}
}
}

使用接口测试工具插入数据:http://124.221.23.47:9200/course_publish/_doc/2

{
"id": 2,
"companyId": 1232141425,
"companyName": "Test1",
"name": "测试课程01",
"users": "IT爱好者IT爱好者IT爱好者IT爱好者IT爱好者IT爱好者IT爱好者IT爱好者",
"tags": "课程标签",
"mt": "1-1",
"st": "1-1-1",
"grade": "204001",
"teachmode": "200002",
"description": "测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程测试课程",
"pic": "/mediafiles/2023/08/23/d13fc2687d9335eddd4dc23ccd8c8b8e.jpg",
"createDate": "2019-09-04 08:49:26",
"status": "203001",
"remark": "没有备注",
"mtName": "前端开发",
"stName": "HTML/CSS",
"charge": "201001",
"price": 1.0,
"originPrice": null,
"validDays": null
}

数据同步

Elasticsearch和MySQL之间需要数据同步,一般有以下几种方式:

  • 同步调用,即写入本地数据库的同时,(异步)调用更新ELK的同步接口,完成同步。
  • 异步通知,即使用消息中间件,写入数据库的同时发送同步消息,监听同步消息完成数据推送。
  • 监听binlog,使用Cannal等组件,可以监听数据库的binlog,完成数据同步。
  • 任务调度,即将任务写入数据库,调度时从数据库查询同步任务,将数据同步到ELK。

本项目使用任务调度方式,适合实时性要求不高的场景!

搜索Client

在包com.swx.content.client下新建Feign的接口

SearchServiceClient
/**
* 远程调用搜索服务
* 新增索引
*/
@FeignClient(value = "search", fallbackFactory = SearchServiceClientFallbackFactory.class)
public interface SearchServiceClient {

@PostMapping("/search/index/course")
public Boolean add(@RequestBody CourseIndex courseIndex);
}

同一包下创建熔断降级处理类:

SearchServiceClientFallbackFactory
@Slf4j
@Component
public class SearchServiceClientFallbackFactory implements FallbackFactory<SearchServiceClient> {

@Override
public SearchServiceClient create(Throwable throwable) {
return courseIndex -> {
log.debug("远程调用搜索服务创建课程索引接口发生熔断: ", throwable);
return false;
};
}
}

完善调度任务逻辑

找到文章发布任务调度类saveCourseIndex,修改生成静态化页面上传至文件系统逻辑

saveCourseIndex
private final SearchServiceClient searchServiceClient;

/**
* 保存课程索引到Elasticsearch
*
* @param mqMessage 消息
* @param courseId 课程ID
*/
private void saveCourseIndex(MqMessage mqMessage, long courseId) {
Long taskId = mqMessage.getId();
MqMessageService mqMessageService = this.getMqMessageService();
// 任务幂等性处理
int stageTwo = mqMessageService.getStageTwo(taskId);
if (stageTwo > 0) {
log.debug("课程索引已缓存,无需处理");
return;
}
// 查询课程
CoursePublish coursePublish = coursePublishService.getById(courseId);
CourseIndex courseIndex = new CourseIndex();
BeanUtils.copyProperties(coursePublish, courseIndex);
// 远程调用
Boolean add = searchServiceClient.add(courseIndex);
if (!add) {
throw new BizException("添加课程索引失败");
}
// 置状态为已完成
mqMessageService.completedStageTwo(taskId);
}

缓存至Redis

测试

测试流程说明

  1. 启动前端的管理项目;

  2. 将一门课程提交审核,打开数据库管理工具,找到course_basecourse_publish_pre两张表,修改其中的status字段为202004,即审核通过;

  3. 再次来到管理项目,将审核通过的课程发布;

  4. 开启任务调度,置状态为RUNNING,等待调度完成;

  5. 打开门户网站,进入课程搜索页面,看到刚发布的课程,表示测试通过。

获取课程

获取已发布课程的详细信息

接口信息如下

路径地址 http://localhost:63040/content/course/whole/2
请求方式 GET
请求参数 Long
返回结果 CoursePreviewVO

查询Service

com.swx.content.service包下的 CoursePublishService 类中定义查询方法

CoursePublishService
/**
* 查询已发布课程的信息
*
* @param courseId 课程ID
* @return com.swx.content.model.po.CoursePublish 课程发布信息
*/
public CoursePreviewVO getCoursePublish(Long courseId);

实现该方法

CoursePublishServiceImpl
/**
* 查询已发布课程的信息
*
* @param courseId 课程ID
* @return com.swx.content.model.po.CoursePublish 课程发布信息
*/
@Override
public CoursePreviewVO getCoursePublish(Long courseId) {
CoursePublish coursePublish = getById(courseId);
if (coursePublish == null) {
throw new BizException("课程未发布");
}
CoursePreviewVO coursePreviewVO = new CoursePreviewVO();

// 解析课程基本信息和营销信息
CourseBaseInfoVO courseBaseInfoVO = new CourseBaseInfoVO();
BeanUtils.copyProperties(coursePublish, courseBaseInfoVO);
coursePreviewVO.setCourseBase(courseBaseInfoVO);

// 解析课程计划信息
List<TeachPlanVO> teachPlans = JSON.parseArray(coursePublish.getTeachplan(), TeachPlanVO.class);
coursePreviewVO.setTeachplans(teachPlans);

return coursePreviewVO;
}

查询Controller

CoursePublishController
@ApiOperation("获取课程发布信息")
@GetMapping("/course/whole/{courseId}")
public CoursePreviewVO getCOursePublish(@PathVariable("courseId") @NotNull(message = "课程ID不能为空") Long courseId) {
return coursePublishService.getCoursePublish(courseId);
}