厚积薄发打卡Day91: MongoDB + RabbitMQ 构建消息推送功能（中）＜MongoDB消息对象＞_通过rabbitmq 推送到mongodb

MongoDB集合：

1. Message消息对象

   集合有什么字段，取决于保存在其中的数据。

   下面这张表格是Message集合中JSON数据的结构要求：

   字段	类型	备注
   _id	UUID	自动生成的主键值
   uuid	UUID	UUID值，并且设置有唯一性索引，防止消息被重复消费
   senderId	Integer	发送者ID，就是用户ID。如果是系统自动发出，这个ID值是0
   senderPhoto	String	发送者的头像URL。在消息页面要显示发送人的头像
   senderName	String	发送者名称，也就是用户姓名。在消息页面要显示发送人的名字
   msg	String	消息正文
   sendTime	Date	发送时间

2. Message_ref消息接收对象

   虽然message集合记录的是消息，里面有接受者ID，但是如果是群发消息，那么接受者ID是空值。这时候就需要用上message_ref集合来记录接收人和已读状态。

   字段	类型	备注
   _id	UUID	主键
   messageId	UUID	message记录的_id
   receiverId	String	接收人ID
   readFlag	Boolean	是否已读
   lastFlag	Boolean	是否为新接收的消息

3. 执行两个集合的联合查询，根据接收人来查询消息，并且按照消息发送时间降序排列，查询前50条记录

   db.message.aggregate([
   	{
   		$set: {
   			"id": { $toString: "$_id" }
   		}
   	},
   	{
   		$lookup:{
   			from:"message_ref",
   			localField:"id",
   			foreignField:"messageId",
   			as:"ref"
   		},
   	},
   	{ $match:{"ref.receiverId": 1} },
   	{ $sort: {sendTime : -1} },
   	{ $skip: 0 },
   	{ $limit: 50 }
   ])
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19

POJO映射类：

1. 在Springboot工程中导入maven依赖：

   <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-data-mongodb</artifactId>
   </dependency>
   1
   2
   3
   4

2. Spring配置：

   spring:
     data:
       mongodb:
         host: #服务器
         port: #端口
         database: #集合
         authentication-database: #用户名
         username: #用户名
         password: #密码
   1
   2
   3
   4
   5
   6
   7
   8
   9

3. spring-data-mongodb主要有以下注解：

   1. @Id：

      主键，不可重复，自带索引，可以在定义的列名上标注，需要自己生成并维护不重复的约束。如果自己不设置@Id主键，mongo会自动生成一个唯一主键，并且插入时效率远高于自己设置主键。

   2. @Document

      标注在实体类上，类似于hibernate的entity注解，标明由mongo来维护该表。

   3. @Indexed

      声明该字段需要加索引，加索引后以该字段为条件检索将大大提高速度。
      唯一索引的话是@Indexed(unique = true)。

4. MessageEntity（Message实体类）

   @Data
   @Document(collection = "message")
   public class MessageEntity implements Serializable {
       @Id
       private String _id;
      
       @Indexed(unique = true)
       private String uuid;
      
       @Indexed
       private Integer senderId;
       
      private String senderPhoto;
       
      private String senderName;
       
      @Indexed
       private Date sendTime;
      
       private String msg;
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21

5. MessageRefEntity（Message_ref实体类）

   @Document(collection = "message_ref")
   @Data
   public class MessageRefEntity implements Serializable {
       @Id
       private String _id;
   
       @Indexed
       private String messageId;
   
       @Indexed
       private Integer receiverId;
   
       @Indexed
       private Boolean readFlag;
      
      @Indexed
       private Boolean lastFlag;
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18

持久层：

1. 就像使用JDBC操作关系型数据库，RedisTemplate操作redis，对应的mongodb也有其操作对象：MongoTemplate

   // 查询Message文档的全部数据
   mongoTemplate.findAll(MessageEntity.class);
   //查询Message文档id为id的数据
   mongoTemplate.findById(<id>, MessageEntity.class);
   //根据query内的查询条件查询
   mongoTemplate.find(query, MessageEntity.class);
   //修改
   mongoTemplate.upsert(query, update, MessageEntity.class);
   //删除
   mongoTemplate.remove(query, MessageEntity.class);
   //新增：
   mongoTemplate.insert(messageEntity);
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12

   具体api查看官方文档：MongoTemplate

2. MessageDao：

   

   @Repository
   @Slf4j
   public class MessageDao {
   
       @Autowired
       private MongoTemplate mongoTemplate;
   
       /**
        * 存入消息主体
        *
        * @param entity
        * @return
        */
       public String insert(MessageEntity entity) {
           //mongodb里面存储的是gmt时间，需要将当前时间换成格林尼治时间
           //gmt为0点  北京时间为8点，所以存数据数据得-8小时
           Date sendTime = entity.getSendTime();
           sendTime = DateUtil.offset(sendTime, DateField.HOUR, 8);
           entity.setSendTime(sendTime);
           entity = mongoTemplate.insert(entity);
           return entity.get_id();
       }
   
       /**
        * 查询集合中消息及其对应的消息接收对象
        *
        * @param userId
        * @param start
        * @param length
        * @return
        */
       //相当用java代码写：
       //db.message.aggregate([
       // {
       //    $set: {
       //       "id": { $toString: "$_id" }
       //    }
       // },
       // {
       //    $lookup:{
       //       from:"message_ref",
       //       localField:"id",
       //       foreignField:"messageId",
       //       as:"ref"
       //    },
       // },
       // { $match:{"ref.receiverId": 1} },
       // { $sort: {sendTime : -1} },
       // { $skip: 0 },
       // { $limit: 50 }
       //])
       public List<HashMap> searchMessageByPage(int userId, long start, int length) {
           JSONObject json = new JSONObject();
           json.set("$toString", "$_id");
           Aggregation aggregation = Aggregation.newAggregation(
                   Aggregation.addFields().addField("id").withValue(json).build(),
                   Aggregation.lookup("message_ref", "id", "messageId", "ref"),
                   Aggregation.match(Criteria.where("ref.receiverId").is(userId)),
                   Aggregation.sort(Sort.by(Sort.Direction.DESC, "sendTime")),
                   Aggregation.skip(start),
                   Aggregation.limit(length)
           );
           AggregationResults<HashMap> results = mongoTemplate.aggregate(aggregation, "message", HashMap.class);
           List<HashMap> list = results.getMappedResults();
           list.forEach(one -> {
               List<MessageRefEntity> refList = (List<MessageRefEntity>) one.get("ref");
               MessageRefEntity entity = refList.get(0);
               boolean readFlag = entity.getReadFlag();
               String refId = entity.get_id();
               one.remove("ref");
               one.put("readFlag", readFlag);
               one.put("refId", refId);
               one.remove("_id");
               //把格林尼治时间转换成北京时间
               Date sendTime = (Date) one.get("sendTime");
               sendTime = DateUtil.offset(sendTime, DateField.HOUR, -8);
   
               String today = DateUtil.today();
               //如果是今天的消息，只显示发送时间，不需要显示日期
               if (today.equals(DateUtil.date(sendTime).toDateStr())) {
                   one.put("sendTime", DateUtil.format(sendTime, "HH:mm"));
               }
               //如果是以往的消息，只显示日期，不显示发送时间
               else {
                   one.put("sendTime", DateUtil.format(sendTime, "yyyy/MM/dd"));
               }
           });
           return list;
       }
   
       public HashMap searchMessageById(String id){
           HashMap map = mongoTemplate.findById(id, HashMap.class, "message");
           Date sendTime = (Date) map.get("sendTime");
           //将GMT时间换成北京时间进行展示
           sendTime = DateUtil.date(sendTime).offset(DateField.HOUR,-8);
           map.replace("sendTime",DateUtil.format(sendTime,"yyyy-MM-dd HH:mm"));
           return map;
       }
   
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59
   60
   61
   62
   63
   64
   65
   66
   67
   68
   69
   70
   71
   72
   73
   74
   75
   76
   77
   78
   79
   80
   81
   82
   83
   84
   85
   86
   87
   88
   89
   90
   91
   92
   93
   94
   95
   96
   97
   98
   99
   100

3. MessageRefDao：

   

   @Repository
   @Slf4j
   public class MessageRefDao {
   
       @Autowired
       private MongoTemplate mongoTemplate;
   
       /**
        * 插入消息对象
        *
        * @param entity
        * @return
        */
       public String insert(MessageRefEntity entity) {
           entity = mongoTemplate.insert(entity);
           return entity.get_id();
       }
   
       /**
        * 查询未读消息
        *
        * @param userId
        * @return
        */
       public long searchUnreadCount(int userId) {
           Query query = new Query();
           query.addCriteria(Criteria.where("readFlag").is(false).and("receiverId").is(userId));
           long count = mongoTemplate.count(query, MessageRefEntity.class);
           return count;
       }
   
       /**
        * 查询新接收消息数量
        *
        * @return
        */
       public long searchLastCount(int userId) {
           Query query = new Query();
           query.addCriteria(Criteria.where("lastFlag").is(true).and("receiverId").is(userId));
           Update update = new Update();
           update.set("lastFlag", false);
           UpdateResult result = mongoTemplate.updateMulti(query, update, "message_ref");
           long rows = result.getModifiedCount();
           log.debug("※此时修改了" + rows + "条message_ref记录");
           return rows;
       }
   
       public long deleteMessageRefById(String id){
           Query query = new Query();
           query.addCriteria(Criteria.where("_id").is(id));
           DeleteResult result = mongoTemplate.remove(query, "message_ref");
           long rows = result.getDeletedCount();
           log.debug("※此时删除了" + rows + "条message_ref记录");
           return rows;
       }
   
       /**
        * 删除某个用户的全部消息
        * @param userId
        * @return
        */
       public long deleteUserMessageRef(int userId){
           Query query = new Query();
           query.addCriteria(Criteria.where("receiverId").is(userId));
           DeleteResult result = mongoTemplate.remove(query, "message_ref");
           long rows = result.getDeletedCount();
           log.debug("※此时一共删除了" + rows + "条message_ref记录");
           return rows;
       }
   
       /**
        * 把未读消息变已读
        * @param id
        * @return
        */
       public long updateUnreadMessage(String id) {
           Query query = new Query();
           query.addCriteria(Criteria.where("_id").is(id));
           Update update = new Update();
           update.set("readFlag", true);
           UpdateResult result = mongoTemplate.updateFirst(query, update, "message_ref");
           long rows = result.getModifiedCount();
           log.debug("※此时查阅了" + rows + "条message_ref记录");
           return rows;
       }
   }
   
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59
   60
   61
   62
   63
   64
   65
   66
   67
   68
   69
   70
   71
   72
   73
   74
   75
   76
   77
   78
   79
   80
   81
   82
   83
   84
   85
   86
   87