tags

src/main/java/com/smcic/api/tags/dto/Recommend.java

@@ -0,0 +1,45 @@
+package com.smcic.api.tags.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@NoArgsConstructor
+@Data
+public class Recommend {
+
+    @JsonProperty("appid")
+    private String appid;
+    @JsonProperty("table_name")
+    private String tableName;
+    @JsonProperty("table_content")
+    private List<TableContentDTO> tableContent;
+
+    @NoArgsConstructor
+    @Data
+    public static class TableContentDTO {
+        @JsonProperty("cmd")
+        private String cmd;
+        @JsonProperty("fields")
+        private FieldsDTO fields;
+
+        @NoArgsConstructor
+        @Data
+        public static class FieldsDTO {
+            @JsonProperty("userid")
+            private String userid;
+            @JsonProperty("tel")
+            private String tel;
+            @JsonProperty("user_tags")
+            private String userTags;
+            @JsonProperty("province")
+            private String province;
+            @JsonProperty("city")
+            private String city;
+            @JsonProperty("imei")
+            private String imei;
+        }
+    }
+}

src/main/java/com/smcic/api/tags/dto/UserTagsDTO.java

@@ -0,0 +1,32 @@
+package com.smcic.api.tags.dto;
+
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.smcic.api.tags.entity.UserTags;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Data
+public class UserTagsDTO {
+
+    @TableId
+    private String uuid;
+
+    private String sex;
+
+    private String ageRange;
+
+    private String province;
+
+    private String city;
+
+    private List<String> tags;
+
+    private String userId;
+
+    private LocalDateTime createTime;
+
+
+
+}

src/main/java/com/smcic/api/tags/entity/UserTags.java

@@ -0,0 +1,46 @@
+package com.smcic.api.tags.entity;
+
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.smcic.api.tags.dto.UserTagsDTO;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Data
+public class UserTags {
+
+    @TableId
+    private String uuid;
+
+    private String sex;
+
+    private String ageRange;
+
+    private String province;
+
+    private String city;
+
+    private String tags;
+
+    private String userId;
+
+    private LocalDateTime createTime;
+
+    public UserTags() {
+
+    }
+
+    public UserTags(UserTagsDTO userTagsDTO) {
+        this.setUuid(userTagsDTO.getUuid());
+        this.setSex(userTagsDTO.getSex());
+        this.setAgeRange(userTagsDTO.getAgeRange());
+        this.setProvince(userTagsDTO.getProvince());
+        this.setCity(userTagsDTO.getCity());
+        this.setTags(String.join(";", userTagsDTO.getTags()));
+        this.setUserId(userTagsDTO.getUserId());
+        this.setCreateTime(userTagsDTO.getCreateTime());
+
+    }
+
+}

src/main/java/com/smcic/api/tags/job/UserTagsConsumer.java

@@ -0,0 +1,100 @@
+package com.smcic.api.tags.job;
+
+import com.smcic.api.tags.dto.Recommend;
+import com.smcic.api.tags.dto.UserTagsDTO;
+import com.smcic.api.tags.entity.UserTags;
+import com.smcic.api.tags.mapper.UserTagsMapper;
+import com.smcic.api.tags.service.TagsQueue;
+import com.smcic.api.tags.service.impl.UserTagsService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.util.TextUtils;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Service;
+import org.springframework.web.client.RestTemplate;
+
+import javax.annotation.Resource;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.stream.Collectors;
+
+@Service
+@Slf4j
+public class UserTagsConsumer {
+
+    @Resource
+    private UserTagsService userTagsService;
+
+    @Resource
+    private UserTagsMapper userTagsMapper;
+
+    @Resource
+    private ThreadPoolExecutor threadPoolExecutor;
+
+
+    public void run(){
+
+        System.out.println("启动队列消费");
+        ConcurrentLinkedDeque<UserTagsDTO> userTagsQueue = TagsQueue.getInstance().getUserTags();
+        long time = System.currentTimeMillis();
+        int count = 0;
+        List<UserTags> userTagsList = new ArrayList<>();
+
+        Recommend recommend = new Recommend();
+        recommend.setAppid("51000");
+        recommend.setTableName("user");
+
+        List<Recommend.TableContentDTO> tableContent = new ArrayList<>();
+
+        while (true){
+            try {
+                UserTagsDTO userTags = userTagsQueue.poll();
+                if (null != userTags){
+                    log.info("接收到数据{}", userTags);
+
+                    userTagsList.add(new UserTags(userTags));
+
+                    Recommend.TableContentDTO tableContentDTO = new Recommend.TableContentDTO();
+                    tableContentDTO.setCmd("update");
+
+                    Recommend.TableContentDTO.FieldsDTO fieldsDTO = new Recommend.TableContentDTO.FieldsDTO();
+//                    fieldsDTO.setUserid(userTags.getUserId());
+                    //fieldsDTO.setTel();
+                    if (!TextUtils.isEmpty(userTags.getAgeRange()))
+                        userTags.getTags().add(userTags.getAgeRange());
+                    fieldsDTO.setUserTags(String.join(";", userTags.getTags()));
+                    fieldsDTO.setProvince(userTags.getProvince());
+                    fieldsDTO.setCity(userTags.getCity());
+
+                    fieldsDTO.setImei(userTags.getUuid());
+
+                    tableContentDTO.setFields(fieldsDTO);
+
+
+                    tableContent.add(tableContentDTO);
+
+                    count++;
+                }
+                long curr = System.currentTimeMillis();
+                if(count > 300 || (count > 0 && curr - time > 2000)){
+                    userTagsMapper.replaceUser(userTagsList);
+                    userTagsList.clear();
+                    count = 0;
+                    time = curr;
+                    this.threadPoolExecutor.execute(() -> {
+                        RestTemplate restTemplate = new RestTemplate();
+                        recommend.setTableContent(tableContent);
+                        ResponseEntity<Object> mapResponseEntity = restTemplate.postForEntity("https://receiver.sxtvs.net/user/qmtssp", recommend, Object.class);
+                        log.info("发送数据{}, 推荐系统响应{}", recommend, mapResponseEntity.getBody());
+                        tableContent.clear();
+                    });
+
+
+                }
+            }catch (Exception e){
+                log.info("队列处理异常", e);
+            }
+        }
+    }
+}

src/main/java/com/smcic/api/tags/mapper/UserTagsMapper.java

@@ -0,0 +1,10 @@
+package com.smcic.api.tags.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.smcic.api.tags.entity.UserTags;
+
+import java.util.List;
+
+public interface UserTagsMapper extends BaseMapper<UserTags> {
+Integer replaceUser(List<UserTags> userTagsList);
+}

src/main/java/com/smcic/api/tags/service/IUserTagsService.java

@@ -0,0 +1,7 @@
+package com.smcic.api.tags.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.smcic.api.tags.entity.UserTags;
+
+public interface IUserTagsService extends IService<UserTags> {
+}

src/main/java/com/smcic/api/tags/service/TagsQueue.java

@@ -0,0 +1,33 @@
+package com.smcic.api.tags.service;
+
+import com.smcic.api.tags.dto.UserTagsDTO;
+import com.smcic.api.tags.entity.UserTags;
+
+import java.util.concurrent.ConcurrentLinkedDeque;
+
+public class TagsQueue {
+    private static volatile TagsQueue tagsQueue;
+
+    private final ConcurrentLinkedDeque<UserTagsDTO> userTags;
+
+    private TagsQueue(){
+        this.userTags = new ConcurrentLinkedDeque<>();
+    }
+
+    public static TagsQueue getInstance(){
+        if (tagsQueue == null){
+            synchronized (TagsQueue.class){
+                if (tagsQueue == null){
+                    tagsQueue = new TagsQueue();
+                }
+            }
+        }
+        return tagsQueue;
+    }
+
+    public ConcurrentLinkedDeque<UserTagsDTO> getUserTags(){
+        return this.userTags;
+    }
+
+
+}

src/main/java/com/smcic/api/tags/service/impl/UserTagsService.java

@@ -0,0 +1,39 @@
+package com.smcic.api.tags.service.impl;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.smcic.api.tags.dto.UserTagsDTO;
+import com.smcic.api.tags.entity.UserTags;
+import com.smcic.api.tags.job.UserTagsConsumer;
+import com.smcic.api.tags.mapper.UserTagsMapper;
+import com.smcic.api.tags.service.IUserTagsService;
+import com.smcic.api.tags.service.TagsQueue;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+import java.time.LocalDateTime;
+import java.util.concurrent.ThreadPoolExecutor;
+
+@Service
+@Slf4j
+public class UserTagsService extends ServiceImpl<UserTagsMapper, UserTags> implements IUserTagsService {
+
+    @Resource
+    private UserTagsConsumer userTagsConsumer;
+
+    @Resource
+    private ThreadPoolExecutor threadPoolExecutor;
+
+    @PostConstruct
+    public void init(){
+        threadPoolExecutor.execute(() -> {
+            userTagsConsumer.run();
+        });
+    }
+    public void insert(UserTagsDTO userTags){
+        log.info("insert userTags:{}",userTags);
+        userTags.setCreateTime(LocalDateTime.now());
+        TagsQueue.getInstance().getUserTags().add(userTags);
+    }
+}