上篇寫到,將設備數據改傳到monggodb,本篇博文記錄一下具體過程。
SpringBoot內置Mongdb模塊MongoTemplate,類似於RedisTemplate
1.添加依賴
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-mongodb</artifactId> </dependency>
2.增加配置
spring: data: mongodb: uri: mongodb://username:password@ip:posrt/database
3.配置連接信息(這里簡單連接,如果需要配置連接池等信息,可自行查看源碼,在URL后拼接參數)
com.mongodb.MongoClientSettings.Builder#applyConnectionString

private static final String MONGODB_PREFIX = "mongodb://"; private static final String MONGODB_SRV_PREFIX = "mongodb+srv://"; private static final Set<String> ALLOWED_OPTIONS_IN_TXT_RECORD = new HashSet<String>(asList("authsource", "replicaset")); private static final String UTF_8 = "UTF-8"; private static final Logger LOGGER = Loggers.getLogger("uri"); private final MongoCredential credential; private final boolean isSrvProtocol; private final List<String> hosts; private final String database; private final String collection; private final String connectionString; private ReadPreference readPreference; private WriteConcern writeConcern; private Boolean retryWrites; private Boolean retryReads; private ReadConcern readConcern; private Integer minConnectionPoolSize; private Integer maxConnectionPoolSize; private Integer maxWaitTime; private Integer maxConnectionIdleTime; private Integer maxConnectionLifeTime; private Integer connectTimeout; private Integer socketTimeout; private Boolean sslEnabled; private Boolean sslInvalidHostnameAllowed; private String requiredReplicaSetName; private Integer serverSelectionTimeout; private Integer localThreshold; private Integer heartbeatFrequency; private String applicationName; private List<MongoCompressor> compressorList; private UuidRepresentation uuidRepresentation; 4.去除_class,添加自動索引 /** * @program: * @description: 芒果DBhepper配置項 * @Author: Zhangyb * @CreateDate: 15:37 * @UpdateUser: * @UpdateDate * @UpdateRemark: * @Version: 1.0 */ @Configuration //@ComponentScan(basePackages = {"com.bysk.base.mongodb"}) // 聚合工程, public class MongoConfig { @Autowired private MongoDatabaseFactory mongoDatabaseFactory; @Autowired private MongoMappingContext mongoMappingContext; @Bean public MappingMongoConverter mappingMongoConverter() { mongoMappingContext.setAutoIndexCreation(true); mongoMappingContext.afterPropertiesSet(); DbRefResolver dbRefResolver = new DefaultDbRefResolver(mongoDatabaseFactory); MappingMongoConverter converter = new MappingMongoConverter(dbRefResolver, mongoMappingContext); // 此處是去除插入數據庫的 _class 字段 converter.setTypeMapper(new DefaultMongoTypeMapper(null)); return converter; } }
4.去除_class,添加自動索引

/** * @program: * @description: 芒果DBhepper配置項 * @Author: Zhangyb * @CreateDate: 15:37 * @UpdateUser: * @UpdateDate * @UpdateRemark: * @Version: 1.0 */ @Configuration //@ComponentScan(basePackages = {"com.bysk.base.mongodb"}) // 聚合工程, public class MongoConfig { @Autowired private MongoDatabaseFactory mongoDatabaseFactory; @Autowired private MongoMappingContext mongoMappingContext; @Bean public MappingMongoConverter mappingMongoConverter() { mongoMappingContext.setAutoIndexCreation(true); mongoMappingContext.afterPropertiesSet(); DbRefResolver dbRefResolver = new DefaultDbRefResolver(mongoDatabaseFactory); MappingMongoConverter converter = new MappingMongoConverter(dbRefResolver, mongoMappingContext); // 此處是去除插入數據庫的 _class 字段 converter.setTypeMapper(new DefaultMongoTypeMapper(null)); return converter; } }
5.創建集合實體類
/** * 設備數據 * * @author Mark sunlightcs@gmail.com * @since 1.0.0 2020-11-06 */ @Data @Document(value="device_monitor_record") @ToString public class MgdbDeviceMonitorRecord { @Id String id; @ApiModelProperty(value = "mqtt上傳的設備序列號") private String equipmentId; private String monitorTime; @ApiModelProperty("空氣濕度") private BigDecimal airHumidity; @ApiModelProperty("大氣溫度") private BigDecimal airTemperature; }
6.分頁對象

import java.io.Serializable; import java.util.List; /** * @program: bysk * @Description: mongodb 分頁對象 * @Author: johnny * @CreateDate: 2021/1/8 16:07 * @UpdateUser: 更新者 * @UpdateDate: 2021/1/8 16:07 * @UpdateRemark: 更新說明 * @Version: 1.0 */ public class Page<T> implements Serializable { private static final long serialVersionUID = 1L; /** * 總數 */ protected long total = 0; /** * 每頁顯示條數,默認 10 */ protected long size = 10; /** * 當前頁 */ protected long current = 1; /** * 總頁數 */ protected long pages = 1; /** * 結果列表 */ private List<T> rows; public Page(){ this.current = 1; this.size = 10; } public Page(int currentPage, int pageSize){ this.current=currentPage<=0?1:currentPage; this.size=pageSize<=0?1:pageSize; } public long getSize() { return size; } public void setSize(long pageSize) { this.size = pageSize; } public long getCurrent() { return this.current; } public Page<T> setCurrent(long current) { this.current = current; return this; } public long getTotal() { return this.total; } public Page<T> setTotal(long total) { this.total = total; return this; } public void setPages(long pages){ this.pages = pages; } public long getPages(){ return this.pages; } /** * 設置結果 及總頁數 * @param rows */ public void build(List<T> rows) { this.setRows(rows); long count = this.getTotal(); long divisor = count / this.getSize(); long remainder = count % this.getSize(); this.setPages(remainder == 0 ? divisor == 0 ? 1 : divisor : divisor + 1); } public List<T> getRows() { return rows; } public void setRows(List<T> rows) { this.rows = rows; } }
7.CRUDbase接口

import com.bysk.base.mongdb.pojo.Page; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; import org.springframework.stereotype.Service; import java.util.List; /** * @program: bysk * @Description: 作用描述 * @Author: johnny * @CreateDate: 2021/1/8 16:20 * @UpdateUser: 更新者 * @UpdateDate: 2021/1/8 16:20 * @UpdateRemark: 更新說明 * @Version: 1.0 */ public interface IBaseMongoService<T> { /** * 保存一個對象到mongodb * * @param entity * @return */ public T save(T entity) ; /** * 根據id刪除對象 */ public void deleteById(String id); /** * 根據對象的屬性刪除 * @param t */ public void deleteByCondition(T t); /** * 根據id進行更新 * @param id * @param t */ public void updateById(String id, T t); /** * 根據對象的屬性查詢 * @param t * @return */ public List<T> findByCondition(T t); /** * 通過條件查詢實體(集合) * * @param query */ public List<T> find(Query query) ; /** * 通過一定的條件查詢一個實體 * * @param query * @return */ public T findOne(Query query) ; /** * 通過一定的條件查詢一個實體 * * @param t * @return */ public T findOne(T t) ; /** * 通過條件查詢更新數據 * * @param query * @param update * @return */ public void update(Query query, Update update) ; /** * 通過ID獲取記錄 * * @param id * @return */ public T findById(String id) ; /** * 通過ID獲取記錄,並且指定了集合名(表的意思) * * @param id * @param collectionName * 集合名 * @return */ public T findById(String id, String collectionName) ; /** * 通過條件查詢,查詢分頁結果 * @param page * @param query * @return */ public Page<T> findPage(Page<T> page, Query query); public Page<T> findPageByCondition(Page<T> page,T t); /** * 求數據總和 * @param query * @return */ public long count(Query query); /** * 獲取MongoDB模板操作 * @return */ public MongoTemplate getMongoTemplate(); }
8.CRUDbase接口實現(這里mongdb住建名稱默認ID,如果要靈活控制,可根據反射獲取@ID注解設置)

import cn.craccd.mongoHelper.utils.FormatUtils; import cn.hutool.core.convert.Convert; import cn.hutool.core.util.StrUtil; import cn.hutool.json.JSONUtil; import com.bysk.base.mongdb.servicebase.IBaseMongoService; import com.bysk.base.mongdb.pojo.Page; import lombok.extern.slf4j.Slf4j; import org.bson.Document; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.convert.MongoConverter; import org.springframework.data.mongodb.core.convert.QueryMapper; import org.springframework.data.mongodb.core.convert.UpdateMapper; import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; import javax.annotation.PostConstruct; import java.lang.reflect.Field; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.util.List; /** * @program: bysk * @Description: 作用描述 * @Author: johnny * @CreateDate: 2021/1/8 16:21 * @UpdateUser: 更新者 * @UpdateDate: 2021/1/8 16:21 * @UpdateRemark: 更新說明 * @Version: 1.0 */ @Slf4j public abstract class BaseMongoServiceImpl<T> implements IBaseMongoService<T> { @Autowired protected MongoTemplate mongoTemplate; @Autowired MongoConverter mongoConverter; QueryMapper queryMapper; UpdateMapper updateMapper; @Value("${spring.profiles.active}") private String env; private Boolean print; @PostConstruct public void init() { queryMapper = new QueryMapper(mongoConverter); updateMapper = new UpdateMapper(mongoConverter); print = StrUtil.containsAny(env,"dev","test"); print=false; } /** * 保存一個對象到mongodb * @param bean * @return */ @Override public T save(T bean) { logSave(bean); mongoTemplate.save(bean); return bean; } /** * 根據id刪除對象 */ @Override public void deleteById(String id) { Query query = new Query(); query.addCriteria(Criteria.where("id").is(id)); logDelete(query); mongoTemplate.remove(this.findById(id)); } /** * 根據對象的屬性刪除 * @param t */ @Override public void deleteByCondition(T t) { Query query = buildBaseQuery(t); logDelete(query); mongoTemplate.remove(query, getEntityClass()); } /** * 根據id進行更新 * @param id * @param t */ @Override public void updateById(String id, T t) { Query query = new Query(); query.addCriteria(Criteria.where("id").is(id)); Update update = buildBaseUpdate(t); logUpdate(query,update,false); update(query, update); } /** * 根據對象的屬性查詢 * @param t * @return */ @Override public List<T> findByCondition(T t) { Query query = buildBaseQuery(t); logQuery(query); return mongoTemplate.find(query, getEntityClass()); } /** * 通過條件查詢實體(集合) * @param query * @return */ @Override public List<T> find(Query query) { logQuery(query); return mongoTemplate.find(query, this.getEntityClass()); } /** * 通過一定的條件查詢一個實體 * @param query * @return */ @Override public T findOne(Query query) { logQuery(query); return mongoTemplate.findOne(query, this.getEntityClass()); } @Override public T findOne(T t) { Query query = buildBaseQuery(t); return findOne(query); } /** * 通過條件查詢更新數據 * @param query * @param update */ @Override public void update(Query query, Update update) { logUpdate(query,update,false); mongoTemplate.updateMulti(query, update, this.getEntityClass()); } /** * 通過ID獲取記錄 * @param id * @return */ @Override public T findById(String id) { Class<T> entityClass = this.getEntityClass(); logQuery( new Query(Criteria.where("id").is(id))); return mongoTemplate.findById(id, entityClass); } /** * 通過ID獲取記錄,並且指定了集合名(表的意思) * @param id * @param collectionName * @return */ @Override public T findById(String id, String collectionName) { return mongoTemplate.findById(id, this.getEntityClass(), collectionName); } /** * 通過條件查詢,查詢分頁結果 * @param page * @param query * @return */ @Override public Page<T> findPage(Page<T> page, Query query) { //如果沒有條件 則所有全部 query=query==null?new Query(Criteria.where("_id").exists(true)):query; long count = this.count(query); // 總數 page.setTotal(count); long currentPage = page.getCurrent(); long pageSize = page.getSize(); query.skip((currentPage - 1) * pageSize).limit(Convert.toInt(pageSize)); logQuery(query); List<T> rows = this.find(query); page.build(rows); return page; } @Override public Page<T> findPageByCondition(Page<T> page,T t){ Query query = buildBaseQuery(t); return findPage(page,query); } /** * 求數據總和 * @param query * @return */ @Override public long count(Query query){ return mongoTemplate.count(query, this.getEntityClass()); } /** * 根據vo構建查詢條件Query * @param t * @return */ private Query buildBaseQuery(T t) { Query query = new Query(); Field[] fields = t.getClass().getDeclaredFields(); for (Field field : fields) { field.setAccessible(true); try { Object value = field.get(t); if (value != null) { query.addCriteria(Criteria.where(field.getName()).is(value)); } } catch (IllegalArgumentException e) { e.printStackTrace(); } catch (IllegalAccessException e) { e.printStackTrace(); } } return query; } /** * 根據vo構建更新條件Query * @param t * @return */ private Update buildBaseUpdate(T t) { Update update = new Update(); Field[] fields = t.getClass().getDeclaredFields(); for (Field field : fields) { field.setAccessible(true); try { Object value = field.get(t); if (value != null) { update.set(field.getName(), value); } } catch (Exception e) { e.printStackTrace(); } } return update; } /** * 獲取需要操作的實體類class * @return */ @SuppressWarnings("unchecked") protected Class<T> getEntityClass() { return getSuperClassGenricType(getClass(),0); } /** * 獲取MongoDB模板操作 * @return */ @Override public MongoTemplate getMongoTemplate() { return mongoTemplate; } private Class getSuperClassGenricType(final Class clazz, final int index){ Type genType = clazz.getGenericSuperclass(); if (!(genType instanceof ParameterizedType)) { log.warn(clazz.getSimpleName() + "'s superclass not ParameterizedType"); return Object.class; } Type[] params = ((ParameterizedType) genType).getActualTypeArguments(); if (index >= params.length || index < 0) { log.warn("Index: " + index + ", Size of " + clazz.getSimpleName() + "'s Parameterized Type: " + params.length); return Object.class; } if (!(params[index] instanceof Class)) { log.warn(clazz.getSimpleName() + " not set the actual class on superclass generic parameter"); return Object.class; } return (Class) params[index]; } /** * 打印查詢語句 * * @param query */ private void logQuery( Query query) { if (print) { Class<?> clazz = this.getEntityClass(); MongoPersistentEntity<?> entity = mongoConverter.getMappingContext().getPersistentEntity(clazz); Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity); Document mappedField = queryMapper.getMappedObject(query.getFieldsObject(), entity); Document mappedSort = queryMapper.getMappedObject(query.getSortObject(), entity); String logStr = "\ndb." + StrUtil.lowerFirst(clazz.getSimpleName()) + ".find("; logStr += FormatUtils.bson(mappedQuery.toJson()) + ")"; if (!query.getFieldsObject().isEmpty()) { logStr += ".projection("; logStr += FormatUtils.bson(mappedField.toJson()) + ")"; } if (query.isSorted()) { logStr += ".sort("; logStr += FormatUtils.bson(mappedSort.toJson()) + ")"; } if (query.getLimit() != 0l) { logStr += ".limit(" + query.getLimit() + ")"; } if (query.getSkip() != 0l) { logStr += ".skip(" + query.getSkip() + ")"; } logStr += ";"; log.info(logStr); } } /** * 打印查詢語句 * * @param query */ private void logCount( Query query) { if (print) { Class<?> clazz = this.getEntityClass(); MongoPersistentEntity<?> entity = mongoConverter.getMappingContext().getPersistentEntity(clazz); Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity); String logStr = "\ndb." + StrUtil.lowerFirst(clazz.getSimpleName()) + ".find("; logStr += FormatUtils.bson(mappedQuery.toJson()) + ")"; logStr += ".count();"; log.info(logStr); } } /** * 打印查詢語句 * * @param query */ private void logDelete(Query query) { if (print) { Class<?> clazz = this.getEntityClass(); MongoPersistentEntity<?> entity = mongoConverter.getMappingContext().getPersistentEntity(clazz); Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity); String logStr = "\ndb." + StrUtil.lowerFirst(clazz.getSimpleName()) + ".remove("; logStr += FormatUtils.bson(mappedQuery.toJson()) + ")"; logStr += ";"; log.info(logStr); } } /** * 打印查詢語句 * * @param query */ private void logUpdate( Query query, Update update, boolean multi) { if (print) { Class<?> clazz = this.getEntityClass(); MongoPersistentEntity<?> entity = mongoConverter.getMappingContext().getPersistentEntity(clazz); Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity); Document mappedUpdate = updateMapper.getMappedObject(update.getUpdateObject(), entity); String logStr = "\ndb." + StrUtil.lowerFirst(clazz.getSimpleName()) + ".update("; logStr += FormatUtils.bson(mappedQuery.toJson()) + ","; logStr += FormatUtils.bson(mappedUpdate.toJson()) + ","; logStr += FormatUtils.bson("{multi:" + multi + "})"); logStr += ";"; log.info(logStr); } } /** * 打印查詢語句 * * @param object * */ private void logSave(Object object) { if (print) { String logStr = "\ndb." + StrUtil.lowerFirst(object.getClass().getSimpleName()) + ".save("; logStr += JSONUtil.toJsonPrettyStr(object); logStr += ");"; log.info(logStr); } } /** * 打印查詢語句 * */ private void logSave(List<?> list) { if (print && list.size() > 0) { Object object = list.get(0); String logStr = "\ndb." + StrUtil.lowerFirst(object.getClass().getSimpleName()) + ".save("; logStr += JSONUtil.toJsonPrettyStr(list); logStr += ");"; log.info(logStr); } } }
10.仿mybatisplus方式實現單表CRUD接口以及接口實現類(注入接口后可使用基礎方法,也可以使用自定義方法)
import com.bysk.base.mongdb.pojo.MgdbDeviceMonitorRecord; import com.bysk.base.mongdb.servicebase.IBaseMongoService; public interface MgdbDeviceMonitorRecordService extends IBaseMongoService<MgdbDeviceMonitorRecord> { }
import com.bysk.base.mongdb.pojo.MgdbDeviceMonitorRecord; import com.bysk.base.mongdb.service.MgdbDeviceMonitorRecordService; import com.bysk.base.mongdb.servicebase.impl.BaseMongoServiceImpl; import org.springframework.stereotype.Service; @Service public class MgdbDeviceMonitorRecordServiceImpl extends BaseMongoServiceImpl<MgdbDeviceMonitorRecord> implements MgdbDeviceMonitorRecordService { }
11.使用
@Autowired MgdbDeviceMonitorRecordService mgdbDeviceMonitorRecordService; public void saveOrUpdateForMgDb(String equipmentId, DeviceMonitorRecordDTO bean) { //1.取得數據存儲時間 LocalDateTime monitorTime = bean.getMonitorTime(); //2.計算時間偏移量 LocalDateTime frontTime = monitorTime.minusSeconds(dataRangeTime); LocalDateTime backTime = monitorTime.plusSeconds(dataRangeTime); //3.構建查詢條件 Query query = new Query(); query.addCriteria(Criteria.where("equipmentId").is(equipmentId).and("monitorTime") .gte(LocalDateTimeUtil.format(frontTime, DatePattern.NORM_DATETIME_PATTERN)) .lte(LocalDateTimeUtil.format(backTime, DatePattern.NORM_DATETIME_PATTERN))); MgdbDeviceMonitorRecord oneByQuery1 = mgdbDeviceMonitorRecordService.findOne(query); // 5.存儲邏輯 if (oneByQuery1 == null) { // 不存在 保存_ String beanStr = JSONUtil.toJsonStr(bean); MgdbDeviceMonitorRecord mgdbDeviceMonitorRecord = JSONUtil.toBean(beanStr, MgdbDeviceMonitorRecord.class); mgdbDeviceMonitorRecord.setMonitorTime(LocalDateTimeUtil.format(bean.getMonitorTime(), DatePattern.NORM_DATETIME_PATTERN)); mgdbDeviceMonitorRecord.setId(null); mgdbDeviceMonitorRecordService.save(mgdbDeviceMonitorRecord); log.info(monitorTime+"==數據存儲芒果DB成功=="); } else { String beanStr = JSONUtil.toJsonStr(bean); MgdbDeviceMonitorRecord mgdbDeviceMonitorRecord = JSONUtil.toBean(beanStr, MgdbDeviceMonitorRecord.class); mgdbDeviceMonitorRecord.setMonitorTime(LocalDateTimeUtil.format(bean.getMonitorTime(), DatePattern.NORM_DATETIME_PATTERN)); mgdbDeviceMonitorRecord.setId(oneByQuery1.getId()); mgdbDeviceMonitorRecordService.updateById(oneByQuery1.getId(),mgdbDeviceMonitorRecord); log.info(monitorTime+"==數據更新芒果DB成功=="); } }