做暧暖免费观看网站,家在深圳app下载,电子商务网站建设问题,公司网站建设的申请#x1f388; 1 参考文档 RabbitMQ实现数据库与ElasticSearch的数据同步 | Hannya。-CSDN 企业级开发项目实战——基于RabbitMQ实现数据库、elasticsearch的数据同步 | 波总说先赚它一个小目标-CSDN SPringBoot集成RabbitMQ实现30秒过期删除功能 | 军大君-CSDN #x1f50d; … 1 参考文档 RabbitMQ实现数据库与ElasticSearch的数据同步 | Hannya。-CSDN 企业级开发项目实战——基于RabbitMQ实现数据库、elasticsearch的数据同步 | 波总说先赚它一个小目标-CSDN SPringBoot集成RabbitMQ实现30秒过期删除功能 | 军大君-CSDN 2 个人需求 当进行文件上传、文件创建、文件重命名等操作时 通过RabbitMQ 生产者文件服务执行上传、创建、重命名等文件操作将用户文件信息例如文件名、文件ID等发送到RabbitMQ新增队列。消费者查询服务监听RabbitMQ新增队列一旦收到消息将用户文件信息新增或更新到Elasticsearch中。 文件删除时 通过RabbitMQ 生产者文件服务执行文件删除操作将用户文件ID发送到RabbitMQ删除队列。消费者查询服务监听 RabbitMQ 队列一旦收到消息通过用户文件ID从Elasticsearch中删除相应的用户文件信息。 根据文件名进行文件模糊查询 通过OpenFeign 生产者文件服务查询服务调用文件服务提供的OpenFeign接口通过用户文件ID从查询该用户文件是否存在。消费者查询服务如果不存在将数据根据用户文件ID从Elasticsearch中删除。 分享文件时间到期处理 通过RabbitMQ的TTL生存时间 死信队列 生产者文件服务 使用TTL模拟一个“延时队列”在文件分享时间到期后将消息传递到死信队列。消费者文件服务死信监听器监听到之后将分享文件的分享状态改为已过期状态。 3 声明
只是提供思路代码不是很完整直接复制运行不了。
最后面有完整网盘项目代码。 4 OpenFeign相关部分查询服务
4.1 引入依赖
!-- nacos --
dependencygroupIdcom.alibaba.cloud/groupIdartifactIdspring-cloud-starter-alibaba-nacos-discovery/artifactId
/dependency
!-- openfeign --
dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-starter-openfeign/artifactId
/dependency
!-- loadbalancer --
dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-starter-loadbalancer/artifactId
/dependency4.2 application.yml
spring:# nacos注册的服务名application:name: netdisk-searchcloud:nacos:discovery:# 配置注册服务的IP地址server-addr: IP地址:8848username: nacospassword: nacos4.3 FileFeignService 接口
FeignClient(name netdisk-file, configuration FeignInterceptor.class)
public interface FileFeignService {RequestMapping(/file/getUserFile/{userFileId})ResultResponseBoolean getUserFile(PathVariable Long userFileId);
}4.4 EnableFeignClients 注解
ComponentScan(value com.cauli.search.*)
EnableFeignClients(basePackages com.cauli.search)
SpringBootApplication
public class NetdiskSearchApplication {public static void main(String[] args) {SpringApplication.run(NetdiskSearchApplication.class, args);}
}5 Elasticsearch相关部分查询服务
5.1 引入依赖
!-- elasticsearch --
dependencygroupIdco.elastic.clients/groupIdartifactIdelasticsearch-java/artifactIdversion8.0.1/version
/dependency
dependencygroupIdcom.fasterxml.jackson.core/groupIdartifactIdjackson-databind/artifactIdversion2.12.3/version
/dependency
dependencygroupIdjakarta.json/groupIdartifactIdjakarta.json-api/artifactIdversion2.0.1/version
/dependency5.2 application.yml
# elasticsearch相关的配置
elasticsearch:# ES网关地址hostname: IP地址# ES网关端口port: 9200# ES网官方方案scheme: http5.3 ElasticSearchConfig 配置类
Configuration
public class ElasticSearchConfig {Value(${elasticsearch.hostname})String hostname;Value(${elasticsearch.port})int port;Value(${elasticsearch.scheme})String scheme;Beanpublic ElasticsearchClient elasticsearchClient(){// 创建低级客户端RestClient client RestClient.builder(new HttpHost(hostname, port,scheme)).build();// 创建API客户端使用Jackson映射器创建传输层ElasticsearchTransport transport new RestClientTransport(client,new JacksonJsonpMapper());return new ElasticsearchClient(transport);}
} 5.4 Elasticsearch 服务类和服务实现类
public interface ElasticsearchService {/*** 更新ES数据** param fileSearchDTO*/void uploadES(FileSearchDTO fileSearchDTO);/*** 删除ES数据** param userFileId*/void deleteES(Long userFileId);/*** 搜索ES数据** return*/ListSearchFileVO searchES(SearchFileQueryDTO searchFileVO);
}
Slf4j
Service
public class ElasticsearchServiceImpl implements ElasticsearchService {Autowiredprivate ElasticsearchClient elasticsearchClient;Resourceprivate FileFeignService feignService;private final ThreadPoolExecutor executor new ThreadPoolExecutor(12, // 核心线程数20, // 最大线程数1, // 线程存活时间TimeUnit.SECONDS, // 存活时间单位new ArrayBlockingQueue(1000) // 任务队列);public void uploadES(FileSearchDTO fileSearchDTO) {executor.execute(() - {try {elasticsearchClient.index(i - i.index(file_search).id(fileSearchDTO.getUserFileId()).document(fileSearchDTO));} catch (IOException e) {throw new RuntimeException(e);}});}public void deleteES(Long userFileId) {executor.execute(() - {try {elasticsearchClient.delete(d - d.index(file_search).id(String.valueOf(userFileId)));} catch (Exception e) {log.debug(ES删除操作失败请检查配置);}});}Overridepublic ListSearchFileVO searchES(SearchFileQueryDTO searchFileQueryDTO) {int pageNum (int) searchFileQueryDTO.getPageNum() - 1;int pageSize (int) (searchFileQueryDTO.getPageSize() 0 ? 10 : searchFileQueryDTO.getPageSize());SearchResponseFileSearchDTO search null;try {search elasticsearchClient.search(s - s.index(file_search).query(_1 - _1.bool(_2 - _2.must(_3 - _3.bool(_4 - _4.should(_5 - _5.match(_6 - _6.field(fileName).query(searchFileQueryDTO.getFileName()))).should(_5 - _5.wildcard(_6 - _6.field(fileName).wildcard(* searchFileQueryDTO.getFileName() *))))).must(_3 - _3.term(_4 - _4.field(userId).value(StpUtil.getLoginIdAsLong()))))).from(pageNum).size(pageSize).highlight(h - h.fields(fileName, f - f.type(plain).preTags(span classkeyword).postTags(/span)).encoder(HighlighterEncoder.Html)), FileSearchDTO.class);} catch (IOException e) {e.printStackTrace();}ListSearchFileVO searchFileVOList new ArrayList();if (search ! null) {for (HitFileSearchDTO hit : search.hits().hits()) {SearchFileVO searchFileVO new SearchFileVO();BeanUtil.copyProperties(hit.source(), searchFileVO);searchFileVO.setHighLight(hit.highlight());searchFileVOList.add(searchFileVO);// 如果文件不存在也从ES中删除if (!feignService.getUserFile(searchFileVO.getUserFileId()).getData()) {executor.execute(() - this.deleteES(searchFileVO.getUserFileId()));}}}return searchFileVOList;}
}5.5 ElasticsearchController 前端控制器
RestController
RequestMapping(/search)
public class ElasticsearchController {Autowiredprivate ElasticsearchService elasticService;GetMapping(value /searchFile)public RestResultSearchFileVO searchFile(SearchFileQueryDTO searchFileQueryDTO) {ListSearchFileVO searchFileVOList elasticService.searchES(searchFileQueryDTO);return RestResult.success().dataList(searchFileVOList, searchFileVOList.size());}
}5.6 相关实体类
/*** 文件搜索VO*/
Data
public class SearchFileVO {JsonSerialize(using ToStringSerializer.class)private Long userFileId;private String fileName;private String filePath;private String extendName;private Long fileSize;private String fileUrl;private MapString, ListString highLight;private Integer isDir;
}/*** 文件搜索DTO*/
Data
JsonIgnoreProperties(ignoreUnknown true)
public class FileSearchDTO {private String indexName;private String userFileId;private String fileId;private String fileName;private String content;private String fileUrl;private Long fileSize;private Integer storageType;private String identifier;private Long userId;private String filePath;private String extendName;private Integer isDir;private String deleteTime;private String deleteBatchNum;
}/*** 文件查询条件DTO*/
Data
public class SearchFileQueryDTO {ApiModelProperty(文件名)private String fileName;ApiModelProperty(当前页)private long pageNum;ApiModelProperty(每页数量)private long pageSize;
}6 RabbitMQ相关部分
6.1 生产者部分文件服务
6.1.1 引入依赖
!-- nacos --
dependencygroupIdcom.alibaba.cloud/groupIdartifactIdspring-cloud-starter-alibaba-nacos-discovery/artifactId
/dependency
!-- RabbitMQ我的SpringBoot是2.6.8的 --
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency6.1.2 完整application.yml
server:port: 8083
spring:# MySQL配置datasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://127.0.0.1:3306/数据库名?characterEncodingutf8serverTimezoneAsia/ShanghaiuseSSLfalseallowPublicKeyRetrievaltrueusername: rootpassword: MySQL密码# nacos注册的服务名application:name: netdisk-filecloud:nacos:discovery:# 配置注册服务的IP地址server-addr: IP地址:8848username: nacospassword: nacos# rabbitmq相关的配置rabbitmq:host: IP地址port: 5672virtual-host: 虚拟主机名比如/fileusername: 用户名默认guestpassword: 密码默认guest6.1.3 RabbitMQConfig 配置类
Configuration
public class RabbitMQConfig {// 普通交换机public static final String FILE_EXCHANGE file.exchange;// 文件保存相关public static final String QUEUE_FILE_SAVE queue.file.save;public static final String KEY_FILE_SAVE key.file.save;// 文件删除相关public static final String QUEUE_FILE_REMOVE queue.file.remove;public static final String KEY_FILE_REMOVE key.file.remove;// 死信相关public static final String DEAD_LETTER_EXCHANGE deadLetter.exchange;public static final String DEAD_LETTER_QUEUE deadLetter.queue;public static final String KEY_FILE_DEAD_LETTER key.file.dead.letter;//延迟队列public static final String DELAY_QUEUE delay.queue;/*** 文件保存队列** return*/Beanpublic Queue queueFileSave() {return new Queue(QUEUE_FILE_SAVE);}/*** 文件删除队列** return*/Beanpublic Queue queueFileRemove() {return new Queue(QUEUE_FILE_REMOVE);}/*** 交换机** return*/Beanpublic TopicExchange topicExchange() {return new TopicExchange(FILE_EXCHANGE);}/*** 绑定文件保存队列到交换机** return*/Beanpublic Binding bindFileSave() {return BindingBuilder.bind(queueFileSave()).to(topicExchange()).with(KEY_FILE_SAVE);}/*** 绑定文件删除队列到交换机** return*/Beanpublic Binding bindFileRemove() {return BindingBuilder.bind(queueFileRemove()).to(topicExchange()).with(KEY_FILE_REMOVE);}/*** 定义延时队列** return*/Beanpublic Queue delayQueue() {//设置死信交换机和路由keyreturn QueueBuilder.durable(DELAY_QUEUE)//如果消息过时则会被投递到当前对应的死信交换机.withArgument(x-dead-letter-exchange, DEAD_LETTER_EXCHANGE)//如果消息过时死信交换机会根据routing-key投递消息到对应的队列.withArgument(x-dead-letter-routing-key, KEY_FILE_DEAD_LETTER).build();}/*** 定义死信交换机** return*/Beanpublic TopicExchange deadLetterExchange() {return new TopicExchange(DEAD_LETTER_EXCHANGE);}/*** 定义死信队列** return*/Beanpublic Queue deadLetterQueue() {return new Queue(DEAD_LETTER_QUEUE);}/*** 绑定死信队列到死信交换机** return*/Beanpublic Binding deadLetterBinding() {return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(KEY_FILE_DEAD_LETTER);}
}6.1.4 FileDealComp 文件逻辑处理组件伪代码
/*** 文件逻辑处理组件*/
Slf4j
Component
public class FileDealComp {Autowiredprivate RabbitTemplate rabbitTemplate;private final ThreadPoolExecutor executor new ThreadPoolExecutor(12, // 核心线程数20, // 最大线程数1, // 线程存活时间TimeUnit.SECONDS, // 存活时间单位new ArrayBlockingQueue(1000) // 任务队列);/*** 更新ES数据** param userFileId*/public void uploadES(Long userFileId) {executor.execute(() - {FileSearchDTO fileSearchDTO new FileSearchDTO();// 通过用户文件ID查询用户文件信息...// 通过文件ID查询文件信息...// 将用户文件信息和文件信息同步到fileSearchDTO对象... // 消息队列更新ESrabbitTemplate.convertAndSend(RabbitMQConfig.FILE_EXCHANGE, RabbitMQConfig.KEY_FILE_SAVE, fileSearchDTO);});}/*** 删除ES数据** param userFileId*/public void deleteES(Long userFileId) {// 消息队列删除ESrabbitTemplate.convertAndSend(RabbitMQConfig.FILE_EXCHANGE, RabbitMQConfig.KEY_FILE_REMOVE, userFileId);}/*** 分享文件过期处理** param shareBatchNum 分享批次号*/public void expiredShareFile(String shareBatchNum) {Share share new Share();// 根据分享批次号获取分享信息...// 将分享信息同步到share对象...// 定义日期格式SimpleDateFormat sdf new SimpleDateFormat(yyyy-MM-dd HH:mm:ss);long differenceInMillis 0;try {// 解析日期字符串为日期对象Date shareDate sdf.parse(share.getShareTime());Date endDate sdf.parse(share.getEndTime());// 计算时间差毫秒数differenceInMillis endDate.getTime() - shareDate.getTime();} catch (ParseException e) {e.printStackTrace();}// 存活时间String expiration Long.toString(differenceInMillis);// 延时队列rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_QUEUE, share, message - {message.getMessageProperties().setExpiration(expiration);return message;});}
}6.1.5 ExpiredShareFileListener 过期的分享文件处理监听器
Slf4j
Component
RabbitListener(queues my-dlx-queue)
public class ExpiredShareFileListener {Autowiredprivate ShareService shareService;// 死信相关public static final String DEAD_LETTER_EXCHANGE deadLetter.exchange;public static final String DEAD_LETTER_QUEUE deadLetter.queue;public static final String KEY_FILE_DEAD_LETTER key.file.dead.letter;RabbitListener(bindings {QueueBinding(key KEY_FILE_DEAD_LETTER,value Queue(value DEAD_LETTER_QUEUE, durable true),exchange Exchange(value DEAD_LETTER_EXCHANGE, type ExchangeTypes.TOPIC, ignoreDeclarationExceptions true))})public void receiveShareMessage(Share share) {log.info(监听到文件过期处理操作{}, share);// 将share的分享状态改为已过期 → 将share的shareStatus由0改为1...log.info(操作完成{}, share);}
}6.2 消费者部分查询服务
6.2.1 引入依赖
!-- RabbitMQ 我的SpringBoot是2.6.8的 --
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency6.2.2 完整application.yml
server:port: 8084
spring:# MySQL配置datasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://127.0.0.1:3306/数据库名?characterEncodingutf8serverTimezoneAsia/ShanghaiuseSSLfalseallowPublicKeyRetrievaltrueusername: rootpassword: MySQL密码# nacos注册的服务名application:name: netdisk-searchcloud:nacos:discovery:# 配置注册服务的IP地址server-addr: IP地址:8848username: nacospassword: nacosmvc:path match:matching-strategy: ant_path_matcherservlet:multipart:enabled: true# 单个文件最大限制max-file-size: 1024MB# 多个文件最大限制max-request-size: 2048MB# rabbitmq相关的配置rabbitmq:host: IP地址port: 5672virtual-host: 虚拟主机名比如/fileusername: 用户名默认guestpassword: 密码默认guest# elasticsearch相关的配置
elasticsearch:# ES网关地址hostname: IP地址# ES网关端口port: 9200# ES网官方方案scheme: http6.2.3 FileMQListener 文件处理消息队列监听
Slf4j
Component
public class FileMQListener {// 普通交换机public static final String FILE_EXCHANGE file.exchange;// 文件保存相关public static final String QUEUE_FILE_SAVE queue.file.save;public static final String KEY_FILE_SAVE key.file.save;// 文件删除相关public static final String QUEUE_FILE_REMOVE queue.file.remove;public static final String KEY_FILE_REMOVE key.file.remove;Autowiredprivate ElasticsearchService elasticsearchService;/*** 监听文件信息添加操作** param fileSearchDTO*/RabbitListener(bindings {QueueBinding(key KEY_FILE_SAVE,value Queue(value QUEUE_FILE_SAVE, durable true),exchange Exchange(value FILE_EXCHANGE, type ExchangeTypes.TOPIC, ignoreDeclarationExceptions true))})public void receiveFileSaveMessage(FileSearchDTO fileSearchDTO) {try {log.info(监听到文件信息添加操作{}, fileSearchDTO);// 更新ES数据elasticsearchService.uploadES(fileSearchDTO);log.info(添加完成{}, fileSearchDTO);} catch (Exception ex) {ex.printStackTrace();}}/*** 监听文件信息删除操作** param userFileId*/RabbitListener(bindings {QueueBinding(key KEY_FILE_REMOVE,value Queue(value QUEUE_FILE_REMOVE, durable true),exchange Exchange(value FILE_EXCHANGE, type ExchangeTypes.TOPIC, ignoreDeclarationExceptions true))})public void receiveFileDeleteMessage(Long userFileId) {try {log.info(监听到文件信息删除操作{}, userFileId);// 删除ES数据elasticsearchService.deleteES(userFileId);log.info(文件信息删除完成{}, userFileId);} catch (Exception ex) {ex.printStackTrace();}}
}7 代码仓库
netdisk-cloud | Gitee
