jiazhizhong/spring-boot-base-demo
1
0
Fork 0
mirror of https://github.com/RemainderTime/spring-boot-base-demo.git synced 2026-03-04 19:20:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into feature/master-payment

Browse Source 
# Conflicts:
#	pom.xml
#	src/main/java/cn/xf/basedemo/service/impl/UserServiceImpl.java
This commit is contained in:
海言
2025-11-26 16:57:02 +08:00
parent e3349d1fe1 27b45cca66
commit 1a66f33bea
12 changed files with 66 additions and 924 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
src/main/java/cn/xf/basedemo/common/model/EsBaseModel.java
View File

@@ -1,42 +0,0 @@
package cn.xf.basedemo.common.model;
import lombok.Data;
/**
* packageName cn.xf.basedemo.common.model
* @author remaindertime
* @className EsModel
* @date 2024/12/10
* @description es基础模型
*/
@Data
public class EsBaseModel<T> {
public EsBaseModel(String indexName, String documentId, T documentModel, Class<T> clazz) {
this.indexName = indexName;
this.documentId = documentId;
this.documentModel = documentModel;
this.clazz = clazz;
}
/**
* 索引名称
*/
private String indexName;
/**
* 文档id
*/
private String documentId;
/**
* 映射对象
*/
private T documentModel;
/**
* 映射对象类对象
*/
private Class<T> clazz;
}

81
src/main/java/cn/xf/basedemo/common/model/EsSearchModel.java
View File

@@ -1,81 +0,0 @@
package cn.xf.basedemo.common.model;
import lombok.Data;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* packageName cn.xf.basedemo.common.model
* @author remaindertime
* @className EsSearchModel
* @date 2024/12/11
* @description es 搜索模型
*/
@Data
public class EsSearchModel<T> {
public EsSearchModel() {
// 使用 LinkedHashMap 保持插入顺序
this.sort = new LinkedHashMap<>();
}
/**
* 索引名称
*/
private String indexName;
/**
* 文档类型
*/
private Class<T> clazz;
/**
* 页数
*/
private Integer pageNum;
/**
* 每页数量
*/
private Integer pageSize;
/**
* 精准查询字段
*/
private Map<String, Object> termQuery;
/**
* 模糊查询字段(一般是text类型)
*/
private Map<String, Object> matchQuery;
/**
* 排序字段规则 ({"age":"desc"})
*/
private Map<String, String> sort;
/**
* 分组去重字段（支持的字段类型：keyword、numeric、date 和 boolean ）
*/
private String repeatField;;
/**
* 分组嵌套查询别名
*/
private String innerAlias;
/**
* 分组嵌套查询数量
*/
private Integer innerSize;
/**
* 指定需要返回的字段
*/
private List<String> includes;
/**
* 指定需要排除的字段
*/
private List<String> excludes;
}

499
src/main/java/cn/xf/basedemo/common/utils/EsUtil.java
View File

@@ -1,499 +0,0 @@
package cn.xf.basedemo.common.utils;
import cn.xf.basedemo.common.model.EsBaseModel;
import cn.xf.basedemo.common.model.EsSearchModel;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.*;
import co.elastic.clients.elasticsearch._types.query_dsl.*;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.search.*;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.ExistsRequest;
import co.elastic.clients.json.JsonData;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.*;
import java.util.stream.Collectors;
/**
* packageName cn.xf.basedemo.common.utils
* @author remaindertime
* @className EsUtil
* @date 2024/12/10
* @description elasticsearch工具类
*/
@Slf4j
@Component
public class EsUtil {
public static ElasticsearchClient esClient;
{
esClient = (ElasticsearchClient) ApplicationContextUtils.getBean("elasticsearchClient");
}
/**
* 判断索引是否存在
* @param indexName
* @return
*/
public static boolean existIndex(String indexName) {
try {
// 创建 ExistsRequest 请求
ExistsRequest request = new ExistsRequest.Builder()
.index(indexName)
.build();
// 发送请求并获取响应
BooleanResponse response = esClient.indices().exists(request);
// 返回索引是否存在
return response.value();
} catch (Exception e) {
// 处理异常
e.printStackTrace();
return false;
}
}
/**
* 删除索引
*
* @param indexName
*/
@SneakyThrows
public static void delIndex(String indexName) {
if (existIndex(indexName)) {
return;
}
esClient.indices().delete(d -> d.index(indexName));
}
/**
* 创建索引
*
* @param indexName
* @return
*/
public static void createIndex(String indexName) {
if (existIndex(indexName)) {
throw new RuntimeException("索引已经存在");
}
try {
CreateIndexResponse createIndexResponse = esClient.indices().create(c -> c.index(indexName));
// 处理响应
if (createIndexResponse.acknowledged()) {
log.info(" indexed create successfully.");
} else {
log.info("Failed to create index.");
}
} catch (Exception e) {
// 捕获异常并打印详细错误信息
e.printStackTrace();
throw new RuntimeException("创建索引失败，索引名：" + indexName + "，错误信息：" + e.getMessage(), e);
}
}
/**
* 新增文档
* @param esBaseModel
* @return
*/
public static boolean addDocument(EsBaseModel esBaseModel) {
try {
ObjectMapper objectMapper = new ObjectMapper();
String jsonString = objectMapper.writeValueAsString(esBaseModel.getDocumentModel());
log.info("es新增文档，文档内容：{}", jsonString);
// 创建 IndexRequest 实例
IndexRequest request = new IndexRequest.Builder()
.index(esBaseModel.getIndexName())
.id(esBaseModel.getDocumentId()) //指定文档id,不指定会自动生成
.document(esBaseModel.getDocumentModel())
.opType(OpType.Create) // 只会在文档 ID 不存在时创建文档
.build();
IndexResponse response = esClient.index(request);
if ("created".equals(response.result())) {
log.info("Document created: " + response.id());
return true;
} else {
log.info("Document already exists or failed to create.");
return false;
}
} catch (Exception e) {
log.error("es新增文档失败", e);
e.printStackTrace();
}
return false;
}
/**
* 更新文档
* @param esBaseModel
* @return
*/
public boolean updateDocument(EsBaseModel esBaseModel) {
try {
UpdateRequest updateRequest = new UpdateRequest.Builder<>()
.index(esBaseModel.getIndexName())
.id(esBaseModel.getDocumentId())
.doc(esBaseModel.getDocumentModel()).build();
UpdateResponse updateResponse = esClient.update(updateRequest, esBaseModel.getClazz());
log.info("Document updated: " + updateResponse.id());
return true;
} catch (Exception e) {
e.printStackTrace();
}
return false;
}
/**
* 更新文档指定字段（script 脚本）
* @param esBaseModel
* @param script 脚本内容
* @param params 传递参数内容
*/
public void updateDocumentWithScript(EsBaseModel esBaseModel, String script, Map<String, JsonData> params) {
try {
UpdateRequest updateRequest = new UpdateRequest.Builder<>()
.index(esBaseModel.getIndexName())
.id(esBaseModel.getDocumentId())
.script(s ->
s.source(script)// 脚本内容：.source("ctx._source.age += params.increment")
.params(params)) // 传递参数内容：.params("increment",sonData.of(5))
.build();
UpdateResponse updateResponse = esClient.update(updateRequest, esBaseModel.getClazz());
log.info("Document updated: " + updateResponse.id());
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 根据id查询文档
* @param esBaseModel
* @return
*/
public static <T> T getDocumentById(EsBaseModel esBaseModel) {
try {
GetRequest getRequest = new GetRequest.Builder()
.index(esBaseModel.getIndexName())
.id(esBaseModel.getDocumentId())
.build();
GetResponse<T> getResponse = esClient.get(getRequest, esBaseModel.getClazz());
if (getResponse.found()) {
return getResponse.source();
}
} catch (Exception e) {
log.error("es列表查询失败", e);
}
return null;
}
/**
* 查询文档列表
* @param searchModel
* @return
*/
public static <T> List<T> getDocumentList(EsSearchModel searchModel) {
List<T> eslist = new ArrayList<>();
try {
SearchResponse<T> search = esClient.search(buildSearchRequest(searchModel), searchModel.getClazz());
if (Objects.isNull(search)) {
return eslist;
}
HitsMetadata<T> hits = search.hits();
if (Objects.isNull(hits)) {
return eslist;
}
List<Hit<T>> sourceHitList = hits.hits();
if (CollectionUtils.isEmpty(sourceHitList)) {
return eslist;
}
sourceHitList.forEach(item -> {
// 处理每个命中
eslist.add(item.source());
});
return eslist;
} catch (Exception e) {
log.error("es列表查询失败", e);
}
return eslist;
}
/**
* 查询文档数量
* @param searchModel
* @return
*/
public static long getDocumentCount(EsSearchModel searchModel) {
try {
CountRequest.Builder countRequest = new CountRequest.Builder();
countRequest.index(searchModel.getIndexName());
countRequest.query(createBoolQuery(searchModel.getTermQuery(), searchModel.getMatchQuery()));
CountResponse count = esClient.count(countRequest.build());
if (Objects.isNull(count)) {
log.info("es列表数量查询异常{}", searchModel);
return 0;
}
return count.count();
} catch (Exception e) {
log.error("es列表数量查询失败", e);
}
return 0;
}
/**
* 根据id删除文档
* @param esBaseModel
* @return
*/
public static Boolean deleteDocumentById(EsBaseModel esBaseModel) {
try {
DeleteRequest deleteRequest = new DeleteRequest.Builder()
.index(esBaseModel.getDocumentId())
.id(esBaseModel.getDocumentId())
.build();
DeleteResponse deleteResponse = esClient.delete(deleteRequest);
if ("deleted".equals(deleteResponse.result())) {
log.info("Document deleted: " + deleteResponse.id());
return true;
} else {
log.info("Document delete failed: " + deleteResponse.id());
return false;
}
} catch (Exception e) {
log.error("es列表删除失败", e);
}
return false;
}
/**
* 根据条件删除文档
* @param searchModel
* @return 删除数量
*/
public static long deleteDocumentByQuery(EsSearchModel searchModel) {
try {
DeleteByQueryRequest.Builder deleteRequest = new DeleteByQueryRequest.Builder();
deleteRequest.index(searchModel.getIndexName());
deleteRequest.query(createBoolQuery(searchModel.getTermQuery(), searchModel.getMatchQuery()));
deleteRequest.refresh(true); //设置删除操作后是否立即刷新索引，使删除结果立即可见
deleteRequest.timeout(new Time.Builder().time("2s").build()); //设置删除操作的超时时间
deleteRequest.conflicts(Conflicts.Proceed); //Conflicts.Proceed：在版本冲突时继续删除操作;Conflicts.Abort：在版本冲突时中止删除操作
DeleteByQueryResponse dResponse = esClient.deleteByQuery(deleteRequest.build());
if (Objects.nonNull(dResponse)) {
log.info("es条件删除成功，删除数量：{}", dResponse.deleted());
return dResponse.deleted();
}
} catch (Exception e) {
log.error("es条件删除数据失败", e);
}
return 0;
}
/**
* 构建搜索请求对象
* @param searchModel
* @return
*/
private static SearchRequest buildSearchRequest(EsSearchModel searchModel) {
//定义查询对象
SearchRequest.Builder searchRequest = new SearchRequest.Builder();
//设置索引名称
searchRequest.index(searchModel.getIndexName());
//分组去重
if (StringUtils.isNotBlank(searchModel.getRepeatField())) {
searchRequest.collapse(buildCollapse(searchModel));
}
//设置查询条件
searchRequest.query(createBoolQuery(searchModel.getTermQuery(), searchModel.getMatchQuery()));
//设置排序规则
if (searchModel.getSort() != null) {
searchRequest.sort(buildSort(searchModel.getSort()));
}
//设置分页参数
if (searchModel.getPageSize() != null && searchModel.getPageSize() != null) {
searchRequest.from(searchModel.getPageSize() * (searchModel.getPageNum() - 1));
searchRequest.size(searchModel.getPageSize());
}
//设置查询字段/排查字段
SourceConfig sourceConfig = buildSourceConfig(searchModel.getIncludes(), searchModel.getExcludes());
if (Objects.nonNull(sourceConfig)) {
searchRequest.source(sourceConfig);
}
return searchRequest.build();
}
/**
* 构建查询条件
* @param termQuery
* @param matchQuery
* @return
*/
private static Query createBoolQuery(Map<String, Object> termQuery, Map<String, Object> matchQuery) {
BoolQuery.Builder cQuery = new BoolQuery.Builder();
// TermQuery 精准匹配
if (termQuery != null) {
for (Map.Entry<String, Object> entry : termQuery.entrySet()) {
if (Objects.isNull(entry.getValue())) {
continue;
}
String key = entry.getKey();
Object value = entry.getValue();
if (value.getClass().isArray()) { //数组查询，使用 TermsQuery
Object[] values = (Object[]) entry.getValue();
List<FieldValue> objs = Arrays.stream(values)
.map(v -> FieldValue.of(v)) // 将每个对象转换为 FieldValue
.collect(Collectors.toList());
cQuery.must(new TermsQuery.Builder()
.field(key)
.terms(t -> t.value(objs))
.build()
._toQuery());
} else if (value.toString().contains(" ")) { // 短语查询,使用 MatchPhraseQuery (要严格按照单词顺序字符串中有空格，短信需匹配)
cQuery.must(new MatchPhraseQuery.Builder()
.field(key)
.query(value.toString())
.build()
._toQuery());
} else { // 其他情况，使用 TermQuery 精准匹配
cQuery.must(new TermQuery.Builder()
.field(key)
.value(value.toString())
.build()
._toQuery());
}
}
}
// MatchQuery 模糊匹配全文检索分词查询
if (matchQuery != null) {
for (Map.Entry<String, Object> entry : matchQuery.entrySet()) {
if (Objects.isNull(entry.getValue())) {
continue;
}
cQuery.must(new MatchQuery.Builder()
.field(entry.getKey())
.query(entry.getValue().toString())
.build()
._toQuery());
}
}
return cQuery.build()._toQuery();
}
/**
* 构建时间区间查询
* @param startTime 开始时间
* @param endTime 结束时间
* @param fieldName 时间字段
* @return
*/
public static Query createTimeQuery(String startTime, String endTime, String fieldName) {
DateRangeQuery dataQuery = new DateRangeQuery.Builder()
.field(fieldName)
.build();
// 时间区间查询
dataQuery.of(o -> o.gte(startTime));
dataQuery.of(o -> o.lte(endTime));
return dataQuery._toRangeQuery()._toQuery();
}
/**
* 设置查询字段/排查字段
* @param includes 需要字段
* @param excludes 排除字段
* @return
*/
private static SourceConfig buildSourceConfig(List<String> includes, List<String> excludes) {
boolean isIncludes = CollectionUtils.isEmpty(includes);
boolean isExcludes = CollectionUtils.isEmpty(excludes);
//设置查询字段/排查字段
if (isIncludes || isExcludes) {
SourceFilter.Builder sourceFilter = new SourceFilter.Builder();
if (isIncludes)
sourceFilter.includes(includes);
if (isExcludes)
sourceFilter.excludes(excludes);
return new SourceConfig.Builder().filter(sourceFilter.build()).build();
}
return null;
}
/**
* 构建分组去重
* @param searchModel
* @return
*/
private static FieldCollapse buildCollapse(EsSearchModel searchModel) {
FieldCollapse.Builder fieldCollapse = new FieldCollapse.Builder();
//设置分组字段
fieldCollapse.field(searchModel.getRepeatField());
//设置嵌套配置
if (StringUtils.isNotBlank(searchModel.getInnerAlias())) {
InnerHits.Builder innerHits = new InnerHits.Builder();
//设置别名
innerHits.name(searchModel.getInnerAlias());
//设置查询数量
if (searchModel.getInnerSize() != null) {
innerHits.size(searchModel.getInnerSize());
}
fieldCollapse.innerHits(InnerHits.of(i -> i.name(searchModel.getInnerAlias()).size(10)));
}
return fieldCollapse.build();
}
/**
* 构建排序规则
* @param sortMap
* @return
*/
private static List<SortOptions> buildSort(Map<String, String> sortMap) {
if (sortMap == null) {
return null;
}
List<SortOptions> sortList = new ArrayList<>();
for (Map.Entry<String, String> sort : sortMap.entrySet()) {
sortList.add(new SortOptions.Builder().field(f -> f.field(sort.getKey()).order(SortOrder.valueOf(sort.getValue()))).build());
}
return sortList;
}
/**
* 案例：组合多条件查询(关于 must、mustNot、should 条件的使用)
*/
public Query combinationQueryTest() {
//query.must()：and 文档必须满足该条件，如果不满足，文档将不匹配。 and
//query.should()：or 文档可以不满足该条件，但满足该条件时会得分更高；即使不满足，文档也会出现在查询结果中,只是查询结果靠后。
//场景1：文档必须符合所有 must 条件和 mustNot 条件，同时至少满足一个 should 条件。如果 should 条件都不满足，文档将被排除不查询出来。
BoolQuery.Builder query = new BoolQuery.Builder();
//数字范围查询
NumberRangeQuery.Builder numberQuery = new NumberRangeQuery.Builder();
numberQuery.field("age").lte(30.0).build();
// 构建查询条件
query.must(o -> o.term(t -> t.field("status").value("active"))) // 必须满足的条件
.mustNot(o -> o.term(t -> t.field("country").value("China"))) // 不能满足的条件
.filter(f -> f.bool(bo -> bo
.should(so -> so.range(r -> r.number(numberQuery.build()))) // 至少满足一个 should 条件
.should(so -> so.term(t -> t.field("gender").value("male"))) // 至少满足一个 should 条件
.minimumShouldMatch("1") // 至少满足一个 should 条件 也可设置百分比 “50%”
));
//场景2：文档必须符合所有 must 条件和 mustNot 条件，同时至少满足一个 should 条件。如果 should 条件都不满足，不用做额外的过滤（按照should原生特性处理）。
query.must(o -> o.bool(bo -> bo
.should(so -> so.range(r -> r.number(numberQuery.build()))) // 至少满足一个 should 条件
.should(so -> so.term(t -> t.field("gender").value("male"))) // 至少满足一个 should 条件
.minimumShouldMatch("1") // 至少满足一个 should 条件
))
.must(o -> o.term(t -> t.field("status").value("active"))) // 必须满足的条件
.mustNot(o -> o.term(t -> t.field("country").value("China"))); // 不能满足的条件
return query.build()._toQuery();
}
}

79
src/main/java/cn/xf/basedemo/config/EsConfig.java
View File

@@ -1,79 +0,0 @@
package cn.xf.basedemo.config;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
/**
* packageName cn.xf.basedemo.config
* @author remaindertime
* @className ElasticsearchConfig
* @date 2024/12/9
* @description es工具类
*/
@Component
public class EsConfig {
@Value("${elasticsearch.host}")
private String elasticsearchHost;
@Value("${elasticsearch.port}")
private int elasticsearchPort;
@Value("${elasticsearch.username}")
private String username;
@Value("${elasticsearch.password}")
private String password;
/**
-最大连接数 (maxConnTotal)：设置总的最大连接数，取决于业务的并发量。500-2000 之间较为合理。
-每个节点的最大连接数 (maxConnPerRoute)：控制每个节点的最大连接数，建议 50-100 之间。
-IO 线程数 (setIoThreadCount)：根据 CPU 核心数设置，通常为 2-4 倍 CPU 核心数。
-连接超时、套接字超时、获取连接超时：一般设置为 10-30 秒，复杂查询或大数据量操作可适当增加到 20-60 秒。
-失败监听器 (setFailureListener)：自定义重试和故障处理逻辑，确保高可用性。
*/
@Bean
public ElasticsearchClient elasticsearchClient() {
// 创建凭证提供者
CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(
AuthScope.ANY,
new UsernamePasswordCredentials(username, password)
);
// 自定义 RestClientBuilder 配置
RestClientBuilder restClientBuilder = RestClient.builder(
new HttpHost(elasticsearchHost, elasticsearchPort, "http")
).setHttpClientConfigCallback(httpClientBuilder ->
httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider) // 配置认证信息
);
// 配置连接超时、套接字超时、获取连接超时
restClientBuilder.setRequestConfigCallback(builder ->
builder.setConnectTimeout(20000)
.setSocketTimeout(20000)
.setConnectionRequestTimeout(20000)
);
// 创建 RestClientTransport 和 ElasticsearchClient
RestClient restClient = restClientBuilder.build();
ElasticsearchTransport transport = new RestClientTransport(
restClient,
new JacksonJsonpMapper() // 使用 Jackson 进行 JSON 处理
);
return new ElasticsearchClient(transport);
}
/**
window系统本地启动 es8.x 重置密码命令：.\elasticsearch-reset-password -u elastic
*/
}

19
src/main/java/cn/xf/basedemo/controller/business/UserController.java
View File

@@ -39,23 +39,4 @@ public class UserController {
return RetObj.success(loginUser);
}
@Operation(summary = "es同步用户信息", description = "用户信息")
@GetMapping("/syncEs")
public RetObj syncEs(Long userId) {
return userService.syncEs(userId);
}
@Operation(summary = "es查询用户信息", description = "用户信息")
@GetMapping("/getEsId")
public RetObj getEsId(Long userId) {
return userService.getEsId(userId);
}
//发送队列消息
@Operation(summary = "发送队列消息", description = "发送队列消息")
@GetMapping("/sendMsg")
public RetObj sendMsg(String msg) {
return userService.sendMQMsg(msg);
}
}

25
src/main/java/cn/xf/basedemo/mq/RocketMqMsgConsumer.java
View File

@@ -1,25 +0,0 @@
package cn.xf.basedemo.mq;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
* RocketMqMsgComsumer
*
* @author 海言
* @date 2025/10/13
* @time 14:37
* @Description
*/
@Slf4j
@Component
@RocketMQMessageListener(topic = "user-topic",consumerGroup = "consumer-group")
public class RocketMqMsgConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
log.info("接收到消息---------：{}",s);
}
}

39
src/main/java/cn/xf/basedemo/mq/RocketMqMsgProducer.java
View File

@@ -1,39 +0,0 @@
package cn.xf.basedemo.mq;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;
/**
* RocketMqMsgProducer
*
* @author 海言
* @date 2025/10/13
* @time 14:34
* @Description
*/
@Slf4j
@Service
public class RocketMqMsgProducer {
@Resource
private RocketMQTemplate rocketMQTemplate;
//发送普通消息
public void sendMsg(String topic, String msg) {
rocketMQTemplate.convertAndSend(topic, msg);
log.info("发送普通消息：{}", msg);
}
//发送带标签的消息
public void sendMsg(String topic, String tag, String msg) {
rocketMQTemplate.convertAndSend(topic + ":" + tag, msg);
}
//发送延迟消息
public void sendDelayMsg(String topic, String msg, int delayLevel) {
rocketMQTemplate.syncSendDelayTimeMills(topic, msg, delayLevel);
}
}

5
src/main/java/cn/xf/basedemo/service/UserService.java
View File

@@ -14,9 +14,4 @@ public interface UserService {
RetObj login(LoginInfoReq res);
RetObj syncEs(Long userId);
RetObj getEsId(Long userId);
RetObj sendMQMsg(String msg);
}

38
src/main/java/cn/xf/basedemo/service/impl/UserServiceImpl.java
View File

@@ -1,24 +1,19 @@
package cn.xf.basedemo.service.impl;
import cn.xf.basedemo.common.model.EsBaseModel;
import cn.xf.basedemo.common.model.LoginInfo;
import cn.xf.basedemo.common.model.LoginUser;
import cn.xf.basedemo.common.model.RetObj;
import cn.xf.basedemo.common.utils.EsUtil;
import cn.xf.basedemo.common.utils.JwtTokenUtils;
import cn.xf.basedemo.common.utils.RSAUtils;
import cn.xf.basedemo.common.utils.StringUtil;
import cn.xf.basedemo.config.GlobalConfig;
import cn.xf.basedemo.mappers.UserMapper;
import cn.xf.basedemo.model.domain.User;
import cn.xf.basedemo.model.req.LoginInfoReq;
import cn.xf.basedemo.mq.RocketMqMsgProducer;
import cn.xf.basedemo.model.res.LoginInfoRes;
import cn.xf.basedemo.service.UserService;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
@@ -51,9 +46,6 @@ public class UserServiceImpl implements UserService {
@Autowired
private RedisTemplate redisTemplate;
@Resource
private RocketMqMsgProducer rocketMqMsgProducer;
@Override
public RetObj login(LoginInfoReq res) {
@@ -99,32 +91,4 @@ public class UserServiceImpl implements UserService {
return RetObj.success(loginUser);
}
@Override
public RetObj syncEs(Long userId) {
User user = userMapper.selectById(userId);
if (Objects.isNull(user)) {
return RetObj.error("用户不存在");
}
String index = StringUtil.camelToKebabCase(user.getClass().getSimpleName());
if (!EsUtil.existIndex(index)) {
EsUtil.createIndex(index);
}
EsUtil.addDocument(new EsBaseModel(index, String.valueOf(user.getId()), user, user.getClass()));
return RetObj.success();
}
@Override
public RetObj getEsId(Long userId) {
Object user = EsUtil.getDocumentById(new EsBaseModel("user", String.valueOf(userId), null, User.class));
if (Objects.nonNull(user)) {
return RetObj.success(user);
}
return RetObj.error("es中不存在该用户");
}
@Override
public RetObj sendMQMsg(String msg) {
rocketMqMsgProducer.sendMsg("user-topic", msg);
return RetObj.success();
}
}