队列模块代码

applications/school/school-server/src/main/java/com/usoftchina/smartschool/school/business/controller/ScoreController.java

@@ -7,9 +7,7 @@ import com.usoftchina.smartschool.page.PageRequest;
 import com.usoftchina.smartschool.school.business.service.ScoreService;
 import com.usoftchina.smartschool.school.dto.BatchDealBaseDTO;
 import com.usoftchina.smartschool.school.dto.ListReqDTO;
-import com.usoftchina.smartschool.school.po.Notify;
 import com.usoftchina.smartschool.school.po.ScoreForm;
-import com.usoftchina.smartschool.school.po.ScoreImport;
 import com.usoftchina.smartschool.school.po.ScoreImportList;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;

applications/wechat/pom.xml

@@ -14,6 +14,8 @@
     <modules>
         <module>wechat-api</module>
         <module>wechat-server</module>
+        <module>wechat-auth</module>
+        <module>wechat-dto</module>
     </modules>
 
 

applications/wechat/wechat-auth/pom.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>wechat</artifactId>
+        <groupId>com.usoftchina.smartschool</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>wechat-auth</artifactId>
+    <dependencies>
+        <dependency>
+            <groupId>com.usoftchina.smartschool</groupId>
+            <artifactId>core</artifactId>
+        </dependency>
+    </dependencies>
+
+
+</project>

applications/wechat/wechat-auth/src/main/java/com/usoftchina/smartschool/wechat/auth/EnableApiAuthClient.java

@@ -0,0 +1,15 @@
+package com.usoftchina.smartschool.wechat.auth;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * @author: guq
+ * @create: 2019-03-08 09:10
+ **/
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface EnableOpenApiAuthClient {
+}

applications/wechat/wechat-auth/src/main/java/com/usoftchina/smartschool/wechat/auth/configuration/OpenApiConfig.java

@@ -0,0 +1,49 @@
+package com.usoftchina.smartschool.wechat.auth.configuration;
+import org.springframework.beans.factory.annotation.Value;
+/**
+ * @author: guq
+ * @create: 2019-03-08 09:19
+ **/
+public class OpenApiConfig {
+
+    @Value("${openapi.signatureParam:_signature}")
+    private String signatureParam;
+    @Value("${openapi.timestampParam:_timestamp}")
+    private String timestampParam;
+    @Value("${openaip.companyuu:access_id}")
+    private String companyuu;
+    @Value("${openapi.timeout:60000}")
+    private int timeout;
+
+    public String getCompanyuu() {
+        return companyuu;
+    }
+
+    public void setCompanyuu(String companyuu) {
+        this.companyuu = companyuu;
+    }
+
+    public String getSignatureParam() {
+        return signatureParam;
+    }
+
+    public void setSignatureParam(String signatureParam) {
+        this.signatureParam = signatureParam;
+    }
+
+    public String getTimestampParam() {
+        return timestampParam;
+    }
+
+    public void setTimestampParam(String timestampParam) {
+        this.timestampParam = timestampParam;
+    }
+
+    public int getTimeout() {
+        return timeout;
+    }
+
+    public void setTimeout(int timeout) {
+        this.timeout = timeout;
+    }
+}

applications/wechat/wechat-dto/pom.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>wechat</artifactId>
+        <groupId>com.usoftchina.smartschool</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>wechat-dto</artifactId>
+
+
+</project>

applications/wechat/wechat-dto/src/main/java/com/usoftchina/smartschool/wechat/constant/Status.java

@@ -0,0 +1,30 @@
+package com.usoftchina.smartschool.wechat.constant;
+
+/**
+ * @author: guq
+ * @create: 2019-03-07 11:41
+ **/
+public enum Status {
+
+    SEND("已发送"),
+
+    RETRY("重新发送"),
+
+    FAILURE("投递失败"),
+
+    SUCCESS("投递成功");
+
+    private String display;
+
+    public String getDisplay() {
+        return display;
+    }
+
+    public void setDisplay(String display) {
+        this.display = display;
+    }
+
+    Status(String display) {
+        this.display = display;
+    }
+}

applications/wechat/wechat-dto/src/main/java/com/usoftchina/smartschool/wechat/dto/MessageInfoDTO.java

@@ -0,0 +1,152 @@
+package com.usoftchina.smartschool.wechat.dto;
+
+
+import java.io.Serializable;
+
+/**
+ * @author: guq
+ * @create: 2019-03-07 14:05
+ **/
+public class MessageInfoDTO implements Serializable{
+
+    private String msgId;
+    /**
+     * 公众号id
+     */
+    private String appId;
+
+    /**
+     * 公众号密钥
+     */
+    private String secret;
+
+    /*
+     * 推送人微信标识
+     */
+    private String touser;
+
+    /**
+     * 微信模板id
+     */
+    private String templateId;
+
+    private String title;
+
+    private String keyword1;
+
+    private String keyword2;
+
+    private String keyword3;
+
+    private String keyword4;
+
+    private String keyword5;
+
+    private String remark;
+
+    private String url;
+
+    public String getAppId() {
+        return appId;
+    }
+
+    public void setAppId(String appId) {
+        this.appId = appId;
+    }
+
+    public String getSecret() {
+        return secret;
+    }
+
+    public void setSecret(String secret) {
+        this.secret = secret;
+    }
+
+    public String getTouser() {
+        return touser;
+    }
+
+    public void setTouser(String touser) {
+        this.touser = touser;
+    }
+
+    public String getTemplateId() {
+        return templateId;
+    }
+
+    public void setTemplateId(String templateId) {
+        this.templateId = templateId;
+    }
+
+    public String getTitle() {
+        return title;
+    }
+
+    public void setTitle(String title) {
+        this.title = title;
+    }
+
+    public String getKeyword1() {
+        return keyword1;
+    }
+
+    public void setKeyword1(String keyword1) {
+        this.keyword1 = keyword1;
+    }
+
+    public String getKeyword2() {
+        return keyword2;
+    }
+
+    public void setKeyword2(String keyword2) {
+        this.keyword2 = keyword2;
+    }
+
+    public String getKeyword3() {
+        return keyword3;
+    }
+
+    public void setKeyword3(String keyword3) {
+        this.keyword3 = keyword3;
+    }
+
+    public String getKeyword4() {
+        return keyword4;
+    }
+
+    public void setKeyword4(String keyword4) {
+        this.keyword4 = keyword4;
+    }
+
+    public String getRemark() {
+        return remark;
+    }
+
+    public void setRemark(String remark) {
+        this.remark = remark;
+    }
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getKeyword5() {
+        return keyword5;
+    }
+
+    public void setKeyword5(String keyword5) {
+        this.keyword5 = keyword5;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public void setMsgId(String msgId) {
+        this.msgId = msgId;
+    }
+}

applications/wechat/wechat-server/pom.xml

@@ -32,11 +32,6 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
         </dependency>
-        <dependency>
-            <groupId>net.sf.json-lib</groupId>
-            <artifactId>json-lib</artifactId>
-            <classifier>jdk15</classifier>
-        </dependency>
         <dependency>
             <groupId>com.usoftchina.smartschool</groupId>
             <artifactId>server-starter</artifactId>
@@ -45,6 +40,12 @@
         <dependency>
             <groupId>com.usoftchina.smartschool</groupId>
             <artifactId>core</artifactId>
+            <exclusions>
+                <exclusion>
+                    <artifactId>HdrHistogram</artifactId>
+                    <groupId>org.hdrhistogram</groupId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <!--test-->
@@ -64,10 +65,6 @@
             <artifactId>mybatis-spring-boot-starter</artifactId>
         </dependency>
         <!-- sleuth -->
-        <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-zipkin</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.springframework.amqp</groupId>
             <artifactId>spring-rabbit</artifactId>
@@ -90,6 +87,11 @@
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.usoftchina.smartschool</groupId>
+            <artifactId>wechat-dto</artifactId>
+            <version>1.0.0-SNAPSHOT</version>
+        </dependency>
     </dependencies>
 
     <build>

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/WechatApplication.java

@@ -21,28 +21,10 @@ import org.springframework.web.servlet.config.annotation.WebMvcConfigurerAdapter
  **/
 @SpringBootApplication
 @EnableTransactionManagement
-@ServletComponentScan
-@MapperScan("com.usoftchina.smartschool.wechat.mapper")
 @EnableEurekaClient
 @EnableFeignClients("com.usoftchina.smartschool.wechat")
-@ComponentScan(basePackages = {"com.usoftchina.smartschool.wechat"})
-public class WechatApplication  extends WebMvcConfigurerAdapter {
+public class WechatApplication {
     public static void main(String[] args) {
         SpringApplication.run(WechatApplication.class, args);
     }
-
-    //解决跨域问题
-   /* @Override
-    public void addCorsMappings(CorsRegistry registry) {
-        registry.addMapping("/**")
-                .allowedHeaders("*")
-                .allowedOrigins("*")
-                .allowedMethods("*")
-                .allowCredentials(true).maxAge(3600000);
-    }
-
-    @Bean
-    public RestTemplate restTemplate(){
-        return new RestTemplate();
-    }*/
 }

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/config/ExpirationMessagePostProcessor.java

@@ -0,0 +1,26 @@
+package com.usoftchina.smartschool.wechat.config;
+
+
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessagePostProcessor;
+
+/**
+ * @author: guq
+ * @create: 2019-01-07 18:27
+ **/
+public class ExpirationMessagePostProcessor implements MessagePostProcessor {
+
+    private final Long ttl; // 毫秒
+
+    public ExpirationMessagePostProcessor(Long ttl) {
+        this.ttl = ttl;
+    }
+
+    @Override
+    public Message postProcessMessage(Message message) throws AmqpException {
+        message.getMessageProperties() .setExpiration(ttl.toString());
+        // 设置message的失效时间
+        return message;
+    }
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/config/RabbitConfig.java

@@ -0,0 +1,73 @@
+package com.usoftchina.smartschool.wechat.config;
+
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+/**
+ * @author: guq
+ * @create: 2019-03-07 09:31
+ **/
+@ConfigurationProperties(prefix = "rabbit.config")
+public class RabbitConfig {
+
+    private String exchange;
+
+    private String dlexchange;
+
+    private String queue;
+
+    private String dlqueue;
+
+    private String routingkey;
+
+    private Long expired;
+
+    public String getExchange() {
+        return exchange;
+    }
+
+    public void setExchange(String exchange) {
+        this.exchange = exchange;
+    }
+
+    public String getDlexchange() {
+        return dlexchange;
+    }
+
+    public void setDlexchange(String dlexchange) {
+        this.dlexchange = dlexchange;
+    }
+
+    public String getQueue() {
+        return queue;
+    }
+
+    public void setQueue(String queue) {
+        this.queue = queue;
+    }
+
+    public Long getExpired() {
+        return expired;
+    }
+
+    public void setExpired(Long expired) {
+        this.expired = expired;
+    }
+
+    public String getDlqueue() {
+
+        return dlqueue;
+    }
+
+    public void setDlqueue(String dlqueue) {
+        this.dlqueue = dlqueue;
+    }
+
+    public String getRoutingkey() {
+        return routingkey;
+    }
+
+    public void setRoutingkey(String routingkey) {
+        this.routingkey = routingkey;
+    }
+
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/controller/TransfersController.java

@@ -0,0 +1,62 @@
+package com.usoftchina.smartschool.wechat.controller;
+
+
+import com.usoftchina.smartschool.base.Result;
+import com.usoftchina.smartschool.utils.JsonUtils;
+import com.usoftchina.smartschool.wechat.dto.MessageInfoDTO;
+import com.usoftchina.smartschool.wechat.service.SendService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.util.StringUtils;
+import org.springframework.web.bind.annotation.*;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLDecoder;
+import java.util.List;
+
+/**
+ * @author: guq
+ * @create: 2019-01-06 22:15
+ **/
+@RestController
+@RequestMapping("/send")
+//@IgnoreOpenApiAuth
+public class TransfersController {
+
+    @Autowired
+    private SendService sendService;
+
+    /**
+    * @Description 提供系统内部推送使用
+    * @Param: [info]
+    * @return: com.usoftchina.smartschool.base.Result
+    * @Author: guq
+    * @Date: 2019/3/7
+    */
+    @PostMapping("/Message")
+    public Result sendMsg(@RequestBody MessageInfoDTO info) {
+        if (StringUtils.isEmpty(info)) {
+            return Result.error("信息为空");
+        }
+        sendService.sendMessage(info);
+        return Result.success();
+    }
+
+    /**
+    * @Description 对外提供大量数据使用
+    * @Param: [data]
+    * @return: com.usoftchina.smartschool.base.Result
+    * @Author: guq
+    * @Date: 2019/3/7
+    */
+    @PostMapping("/Messages")
+    public Result sendOutMessages(@RequestParam("data")  String data) throws UnsupportedEncodingException {
+        if (StringUtils.isEmpty(data)) {
+            return Result.error("信息为空");
+        }
+        data = URLDecoder.decode(data, "UTF-8");
+        List<MessageInfoDTO> messages = JsonUtils.fromJsonArray(data, MessageInfoDTO.class);
+        //sendService.sendOutMessages(messages);
+        return Result.success();
+    }
+
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/mapper/BrokerMessageLogMapper.java

@@ -0,0 +1,34 @@
+package com.usoftchina.smartschool.wechat.mapper;
+
+
+import com.usoftchina.smartschool.wechat.po.BrokerMessagelog;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * @author: guq
+ * @create: 2019-01-07 20:37
+ **/
+@Mapper
+public interface BrokerMessageLogMapper {
+    BrokerMessagelog getMessage(String msgId);
+
+    BrokerMessagelog getLiveMessage(String msgId);
+
+    BrokerMessagelog getDeadMessage(String msgId);
+
+    int deleteByPrimaryKey(Integer id);
+
+    int insert(BrokerMessagelog record);
+
+    int insertSelective(BrokerMessagelog record);
+
+    BrokerMessagelog selectByPrimaryKey(Integer id);
+
+    int updateByPrimaryKeySelective(BrokerMessagelog record);
+
+    int updateByPrimaryKey(BrokerMessagelog record);
+
+    int updateMessageLogandAddRetry(BrokerMessagelog record);
+
+    void updateSuccess(String msgId);
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/po/BrokerMessagelog.java

@@ -0,0 +1,26 @@
+package com.usoftchina.smartschool.wechat.po;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * @author: guq
+ * @create: 2019-03-07 11:36
+ **/
+@Data
+public class BrokerMessagelog implements Serializable{
+
+    private Long id;
+    private String msgId;
+    private String message;
+    private String status;
+
+    private Date createTime;
+    private int retry;
+    /**
+     * 失败原因
+     */
+    private String reason;
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/po/MessageInfo.java

@@ -0,0 +1,58 @@
+package com.usoftchina.smartschool.wechat.po;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @author: guq
+ * @create: 2019-03-07 14:27
+ **/
+@Data
+public class MessageInfo implements Serializable {
+
+    /**
+     * 消息id
+     */
+    private String msgId;
+    /**
+     * 公众号id
+     */
+    private String appId;
+
+    /**
+     * 公众号密钥
+     */
+    private String secret;
+
+    /*
+     * 推送人微信标识
+     */
+    private String touser;
+
+    /**
+     * 微信模板id
+     */
+    private String templateId;
+
+    private String title;
+
+    private String keyword1;
+
+    private String keyword2;
+
+    private String keyword3;
+
+    private String keyword4;
+
+    private String keyword5;
+
+    private String remark;
+
+    private String url;
+
+    private Boolean send = false;
+
+    //发送失败原因
+    private String reason;
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/po/MessagePackage.java

@@ -0,0 +1,35 @@
+package com.usoftchina.smartschool.wechat.po;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * @author: guq
+ * @create: 2019-03-07 16:10
+ **/
+@Data
+public class MessagePackage implements Serializable{
+
+    private String packageId;
+
+    private List<MessageInfo> data;
+
+    private int retry;
+
+    public MessagePackage(List<MessageInfo> data) {
+        this.packageId = UUID.randomUUID().toString();
+        this.data = data;
+    }
+
+    public MessagePackage(List<MessageInfo> data, String msgId) {
+        this.packageId = msgId;
+        this.data = data;
+    }
+
+    public void addRetry() {
+        this.retry ++;
+    }
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/service/ReceiveService.java

@@ -0,0 +1,95 @@
+package com.usoftchina.smartschool.wechat.service;
+
+import com.rabbitmq.client.Channel;
+
+import com.usoftchina.smartschool.base.Result;
+import com.usoftchina.smartschool.wechat.mapper.BrokerMessageLogMapper;
+import com.usoftchina.smartschool.wechat.po.MessageInfo;
+import com.usoftchina.smartschool.wechat.po.MessagePackage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.annotation.*;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.messaging.handler.annotation.Headers;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author: guq
+ * @create: 2019-01-06 22:04
+ **/
+@Service
+public class ReceiveService {
+
+    @Autowired
+    private SendService sendService;
+
+    @Autowired
+    private WxPushService wxPushService;
+
+
+    @Autowired
+    private BrokerMessageLogMapper brokerMessageLogMapper;
+
+    private Logger logger = LoggerFactory.getLogger(ReceiveService.class);
+
+    private final static String DLEXCHANGE = "saas_trade_dev_dlexchange";
+    private final static String QUEUE = "saas_trade_dev_queue";
+    private final static String ROUTINGKEY = "saas_trade_dev_dl.*";
+
+
+    //配置监听的哪一个队列，同时在没有queue和exchange的情况下会去创建并建立绑定关系
+    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QUEUE,durable = "true"),
+            exchange = @Exchange(name= DLEXCHANGE ,durable = "true",type = "topic"),
+            key = ROUTINGKEY))
+    @RabbitHandler
+    @Transactional
+    public void onMessage(@Payload MessagePackage msgPackage, @Headers Map<String,Object> headers, Channel channel)
+            throws IOException {
+        //消费者操作
+        logger.info("---------收到消息，消息id={},开始消费---------", msgPackage.getPackageId());
+        List<MessageInfo> data = msgPackage.getData();
+        Result result = null;
+        //记录发送失败次数
+        int count = 0;
+        for (MessageInfo msg : data) {
+            if (!msg.getSend()) {
+                result = wxPushService.wechatPush(msg);
+                if (result.isSuccess()) {
+                    msg.setSend(true);
+                } else {
+                    msg.setReason(result.getMessage());
+                    count ++;
+                }
+            }
+        }
+        //存在发送失败的消息
+        if (count > 0) {
+            msgPackage.addRetry();
+            sendService.sendDelayMessage(msgPackage);
+        }
+        /**
+         * Delivery Tag 用来标识信道中投递的消息。RabbitMQ 推送消息给 Consumer 时，会附带一个 Delivery Tag，
+         * 以便 Consumer 可以在消息确认时告诉 RabbitMQ 到底是哪条消息被确认了。
+         * RabbitMQ 保证在每个信道中，每条消息的 Delivery Tag 从 1 开始递增。
+         */
+        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
+        /**
+         *  multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认
+         *  如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认
+         */
+        boolean multiple = false;
+
+        //ACK,确认一条消息已经被消费
+        channel.basicAck(deliveryTag,multiple);
+        logger.info("消息成功消费,MessageInfo={}", msgPackage);
+        //重新返回队列 重新消费
+        //channel.basicNack(deliveryTag,false,true);
+    }
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/service/SendService.java

@@ -0,0 +1,140 @@
+package com.usoftchina.smartschool.wechat.service;
+
+import com.usoftchina.smartschool.exception.BizException;
+import com.usoftchina.smartschool.exception.ExceptionCode;
+import com.usoftchina.smartschool.utils.BeanMapper;
+import com.usoftchina.smartschool.utils.JsonUtils;
+import com.usoftchina.smartschool.wechat.config.ExpirationMessagePostProcessor;
+import com.usoftchina.smartschool.wechat.config.RabbitConfig;
+
+import com.usoftchina.smartschool.wechat.constant.Status;
+import com.usoftchina.smartschool.wechat.dto.MessageInfoDTO;
+import com.usoftchina.smartschool.wechat.mapper.BrokerMessageLogMapper;
+import com.usoftchina.smartschool.wechat.po.BrokerMessagelog;
+import com.usoftchina.smartschool.wechat.po.MessageInfo;
+import com.usoftchina.smartschool.wechat.po.MessagePackage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.support.CorrelationData;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * @author: guq
+ * @create: 2019-01-06 22:04
+ **/
+@Service
+@EnableConfigurationProperties(RabbitConfig.class)
+public class SendService {
+
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    @Autowired
+    private RabbitConfig rabbitConfig;
+
+    @Autowired
+    private BrokerMessageLogMapper brokerMessageLogMapper;
+
+   /* @Value("${smartschool.rabbitmq.retry:1000}")
+    private int retry;*/
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+
+    //发送消息方法调用: 构建自定义对象消息
+    public void sendMessage(MessageInfoDTO messageInfo) {
+        MessageInfo info = BeanMapper.map(messageInfo, MessageInfo.class);
+        if (StringUtils.isEmpty(info.getMsgId())) {
+            info.setMsgId(UUID.randomUUID().toString());
+        }
+        String messageId = info.getMsgId();
+        //统一采用list处理
+        List list = new ArrayList<MessageInfo>();
+        list.add(info);
+        MessagePackage msgPackage = new MessagePackage(list, messageId);
+
+        // 通过实现 ConfirmCallback 接口，消息发送到 Broker 后触发回调，确认消息是否到达 Broker 服务器，也就是只确认是否正确到达 Exchange 中
+        rabbitTemplate.setConfirmCallback(confirmCallback);
+        //消息唯一ID
+        CorrelationData correlationData = new CorrelationData(messageId);
+        rabbitTemplate.convertAndSend(rabbitConfig.getDlexchange(), rabbitConfig.getRoutingkey(), msgPackage, correlationData);
+        logger.info("msgId={}, 消息发送", messageId);
+    }
+
+    //回调函数: confirm确认
+    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
+        @Override
+        public void confirm(CorrelationData correlationData, boolean confirm, String cause) {
+            String messageId = correlationData.getId();
+            if (confirm) {
+                logger.info("msgId={},已发送服务器", messageId);
+            } else {
+                logger.error("msgId={},发送失败:{}", correlationData.getId(), cause);
+            }
+        }
+    };
+
+    //发送消息方法调用: 构建自定义对象消息
+    public void sendDelayMessage(MessagePackage info) {
+        if (StringUtils.isEmpty(info) || StringUtils.isEmpty(info.getPackageId())) {
+            throw new BizException(ExceptionCode.NULL_DATA);
+        }
+        String messageId = info.getPackageId();
+        int count = info.getRetry();
+        //循环次数大于设置值
+        if (count > 12) {
+            saveErrorMessage(info);
+        } else {
+        //未到设置值，继续投放发送至延迟队列
+        rabbitTemplate.setConfirmCallback(confirmCallback);
+        //消息唯一ID
+        CorrelationData correlationData = new CorrelationData(info.getPackageId());
+        rabbitTemplate.convertAndSend(rabbitConfig.getExchange(), rabbitConfig.getRoutingkey(), info,
+                new ExpirationMessagePostProcessor(rabbitConfig.getExpired()), correlationData);
+        }
+    }
+
+    //发送大量数据
+    public void sendOutMessages(List<MessageInfoDTO> messages) {
+        List<MessageInfo> infos = BeanMapper.mapList(messages, MessageInfo.class);
+        MessagePackage msgPackage = new MessagePackage(infos);
+        String messageId = msgPackage.getPackageId();
+
+        //通过实现 ConfirmCallback 接口，消息发送到 Broker 后触发回调，确认消息是否到达 Broker 服务器，也就是只确认是否正确到达 Exchange 中
+        rabbitTemplate.setConfirmCallback(confirmCallback);
+        //消息唯一ID
+        CorrelationData correlationData = new CorrelationData(messageId);
+        rabbitTemplate.convertAndSend(rabbitConfig.getDlexchange(), rabbitConfig.getRoutingkey(), msgPackage, correlationData);
+        logger.info("msgId={}, 消息发送", messageId);
+    }
+
+    private void saveErrorMessage(MessagePackage msgPackage) {
+        String messageId = msgPackage.getPackageId();
+        //消息入库
+        BrokerMessagelog messagelog = new BrokerMessagelog();
+        messagelog.setCreateTime(new Date());
+        messagelog.setMessage(JsonUtils.toJsonString(msgPackage));
+        messagelog.setMsgId(messageId);
+        messagelog.setStatus(Status.FAILURE.getDisplay());
+        brokerMessageLogMapper.insertSelective(messagelog);
+    }
+
+ /*   private void updateErrorMessage(MessagePackage info, String error) {
+        //更新状态且失败次数加1
+        BrokerMessagelog messagelog = new BrokerMessagelog();
+        messagelog.setMsgId(info.getPackageId());
+        messagelog.setStatus(Status.RETRY.getDisplay());
+        messagelog.setReason(error);
+        brokerMessageLogMapper.updateMessageLogandAddRetry(messagelog);
+    }*/
+}

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/service/WxPushService.java

@@ -1,5 +1,7 @@
 package com.usoftchina.smartschool.wechat.service;
 
+import com.usoftchina.smartschool.base.Result;
+import com.usoftchina.smartschool.wechat.po.MessageInfo;
 import org.apache.ibatis.annotations.Param;
 
 public interface WxPushService {
@@ -16,5 +18,6 @@ public interface WxPushService {
                          @Param("keyword3") String keyword3, @Param("keyword4") String keyword4,
                          @Param("remark") String remark, @Param("url") String url);
 
+    public Result wechatPush(MessageInfo info);
     /*public String getWxAccessToken(@Param("appId") String appId, @Param("secret") String secret);*/
 }

applications/wechat/wechat-server/src/main/java/com/usoftchina/smartschool/wechat/service/impl/WxPushServiceImpl.java

@@ -2,8 +2,12 @@ package com.usoftchina.smartschool.wechat.service.impl;
 
 import com.alibaba.fastjson.JSON;
 import com.github.kevinsawicki.http.HttpRequest;
+import com.usoftchina.smartschool.base.Result;
+import com.usoftchina.smartschool.wechat.po.MessageInfo;
 import com.usoftchina.smartschool.wechat.service.WxPushService;
 import com.usoftchina.smartschool.wechat.wxUtils.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.HashMap;
@@ -11,6 +15,10 @@ import java.util.HashMap;
 @Service
 public class WxPushServiceImpl implements WxPushService{
 
+
+    private Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private static final String WECHATURL = "https://api.weixin.qq.com/cgi-bin";
     /**
      * 推送消息
      * @param
@@ -34,11 +42,11 @@ public class WxPushServiceImpl implements WxPushService{
                 params.put("appid", appId);
                 params.put("secret", secret);
                 params.put("grant_type", "client_credential");
-                HttpRequest httpRequest= HttpRequest.get("https://api.weixin.qq.com/cgi-bin/token",params,false);
+                HttpRequest httpRequest= HttpRequest.get(WECHATURL + "/token",params,false);
                 String content=httpRequest.body();
                 System.err.println("getWxAccessTokenContent======"+content);
                 String token= JSON.parseObject(content).getString("access_token");
-                HttpRequest hRequest=  HttpRequest.post("https://api.weixin.qq.com/cgi-bin/message/template/send?access_token="+token)
+                HttpRequest hRequest=  HttpRequest.post(WECHATURL + "/message/template/send?access_token="+token)
                         .header("Content-Type", "application/json")
                         .send(json.getBytes());
                 String result= hRequest.body();
@@ -60,4 +68,42 @@ public class WxPushServiceImpl implements WxPushService{
 
             return token;
         }*/
+
+    /**
+     * 推送消息
+     * @param
+     */
+    @Override
+    public Result wechatPush(MessageInfo info) {
+        if (ObjectUtils.isNotEmpty(info.getTouser())){
+            String json="{\"touser\":\"" + info.getTouser() + "\","
+                    + "\"template_id\":\"" + info.getTemplateId() + "\","
+                    + "\"url\":\"" + "" + info.getUrl() + "" + "\","
+                    + "\"data\":{"
+                    + "\"first\":{\"value\":\"" + info.getTitle() + "\","
+                    +"\"color\":\"#173177\"},"
+                    + "\"keyword1\":{\"value\":\"" + info.getKeyword1() + "\",\"color\":\"#173177\"},"
+                    + "\"keyword2\":{\"value\":\"" + info.getKeyword2() + "\",\"color\":\"#173177\"},"
+                    + "\"keyword3\":{\"value\":\"" + info.getKeyword3() + "\",\"color\":\"#173177\"},"
+                    + "\"keyword4\":{\"value\":\"" + info.getKeyword4() + "\",\"color\":\"#173177\"},"
+                    + "\"remark\":{\"value\":\"" + info.getRemark() + "\",\"color\":\"#173177\"}}}";
+
+            HashMap<String, Object> params=new HashMap<>();
+            params.put("appid", info.getAppId());
+            params.put("secret", info.getSecret());
+            params.put("grant_type", "client_credential");
+            HttpRequest httpRequest= HttpRequest.get(WECHATURL + "/token",params,false);
+            String content=httpRequest.body();
+            String token= JSON.parseObject(content).getString("access_token");
+            HttpRequest hRequest=  HttpRequest.post(WECHATURL + "/message/template/send?access_token="+token)
+                    .header("Content-Type", "application/json")
+                    .send(json.getBytes());
+            String result= hRequest.body();
+            logger.info("result={}", result);
+            if (result.equals("error")) {
+                Result.error(result);
+            }
+        }
+        return Result.success();
+    }
 }

applications/wechat/wechat-server/src/main/resources/application.yml

@@ -22,11 +22,13 @@ spring:
     basename: i18n/messages
     encoding: UTF-8
   rabbitmq:
-    host: 10.10.100.166
-    port: 3306
-    virtual-host: school
-    username: school
-    password: select111***
+    host: 10.10.100.103
+    port: 5672
+    virtual-host: docker
+    username: saas
+    password: select123***
+    publisher-returns: true
+    publisher-confirms: true
   zipkin:
     sender:
       type: rabbit
@@ -89,4 +91,12 @@ hystrix:
                 enabled: true
               isolation:
                     thread:
-                        timeoutInMilliseconds: 4000
+                        timeoutInMilliseconds: 4000
+rabbit:
+  config:
+    exchange: saas_trade_dev_exchange
+    dlexchange: saas_trade_dev_dlexchange
+    queue: saas_trade_dev_queue
+    dlqueue: saas_trade_dev_dlqueue
+    routingkey: saas_trade_dev_dl.*
+    expired: 60000

applications/wechat/wechat-server/src/main/resources/logback-spring.xml

@@ -1,125 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<configuration>
-    <include resource="org/springframework/boot/logging/logback/base.xml" />
-    <jmxConfigurator/>
-
-    <!--
-    %m
-    输出代码中指定的消息
-    %p
-    输出优先级，即DEBUG，INFO，WARN，ERROR，FATAL
-    %r
-    输出自应用启动到输出该log信息耗费的毫秒数
-    %c
-    输出所属的类目，通常就是所在类的全名
-    %t
-    输出产生该日志事件的线程名
-    %n
-    输出一个回车换行符，Windows平台为“\r\n”，Unix平台为“\n”
-    %d
-    输出日志时间点的日期或时间，默认格式为ISO8601，也可以在其后指定格式，比如：%d{yyy MMM dd HH:mm:ss,SSS}，
-    输出类似：2002年10月18日 22：10：28，921
-    %l
-    输出日志事件的发生位置，包括类目名、发生的线程，以及在代码中的行数。举例：Testlog4.main(TestLog4.java:10)
-    -->
-
-    <springProperty scope="context" name="log.path" source="logging.path" defaultValue="/var/log/smartschool/wechat-server"/>
-    <springProperty scope="context" name="spring.application.name" source="spring.application.name" defaultValue="wechat-server"/>
-    <springProperty scope="context" name="spring.profiles.active" source="spring.profiles.active" defaultValue="dev"/>
-    <springProperty scope="context" name="common-pattern" source="logging.common-pattern" defaultValue="%d{yyyy-MM-dd HH:mm:ss.SSS}:[%5p] [%t:%r] [%C{1}:%M:%L] --> %m%n"/>
-    <springProperty scope="context" name="log.level.console" source="logging.level.console" defaultValue="INFO"/>
-    <springProperty scope="context" name="log.destination" source="logging.destination" defaultValue="10.1.81.1:5000"/>
-
-    <contextName>${spring.application.name}-${spring.profiles.active}-logback</contextName>
-
-    <appender name="CONSOLE_APPENDER" class="ch.qos.logback.core.ConsoleAppender">
-        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
-            <level>${log.level.console}</level>
-        </filter>
-        <encoder>
-            <pattern>${common-pattern}</pattern>
-        </encoder>
-    </appender>
-
-    <appender name="ROOT_APPENDER" class="ch.qos.logback.core.rolling.RollingFileAppender">
-        <file>${log.path}/root.log</file>
-        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
-            <fileNamePattern>${log.path}/%d{yyyy-MM}/root-%d{yyyy-MM-dd}-%i.log.gz</fileNamePattern>
-            <maxFileSize>128MB</maxFileSize>
-            <maxHistory>7</maxHistory>
-            <totalSizeCap>20GB</totalSizeCap>
-        </rollingPolicy>
-        <encoder>
-            <pattern>${common-pattern}</pattern>
-        </encoder>
-    </appender>
-
-    <!-- Appender to log in a JSON format -->
-    <appender name="JSON_APPENDER" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
-        <destination>${log.destination}</destination>
-        <encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
-            <providers>
-                <pattern>
-                    <pattern>
-                        {
-                        "severity": "%level",
-                        "service": "${spring.application.name:-}",
-                        "trace": "%X{X-B3-TraceId:-}",
-                        "span": "%X{X-B3-SpanId:-}",
-                        "parent": "%X{X-B3-ParentSpanId:-}",
-                        "exportable": "%X{X-Span-Export:-}",
-                        "pid": "${PID:-}",
-                        "thread": "%thread",
-                        "class": "%logger{40}",
-                        "rest": "%message"
-                        }
-                    </pattern>
-                </pattern>
-            </providers>
-        </encoder>
-    </appender>
-
-    <logger name="org.springframework" level="INFO"/>
-    <logger name="com.usoftchina.smartschool" level="INFO"/>
-
-    <springProfile name="dev">
-        <root level="INFO">
-            <appender-ref ref="CONSOLE_APPENDER"/>
-        </root>
-    </springProfile>
-
-    <springProfile name="test">
-        <root level="INFO">
-            <appender-ref ref="CONSOLE_APPENDER"/>
-            <appender-ref ref="ROOT_APPENDER"/>
-        </root>
-    </springProfile>
-
-    <springProfile name="docker">
-        <logger name="org.springframework" level="WARN"/>
-        <logger name="com.usoftchina.smartschool" level="WARN"/>
-        <root level="WARN">
-            <appender-ref ref="CONSOLE_APPENDER"/>
-            <appender-ref ref="JSON_APPENDER"/>
-        </root>
-    </springProfile>
-
-    <springProfile name="docker-test">
-        <logger name="org.springframework" level="WARN"/>
-        <logger name="com.usoftchina.smartschool" level="WARN"/>
-        <root level="WARN">
-            <appender-ref ref="CONSOLE_APPENDER"/>
-            <appender-ref ref="JSON_APPENDER"/>
-        </root>
-    </springProfile>
-
-    <springProfile name="docker-prod">
-        <logger name="org.springframework" level="WARN"/>
-        <logger name="com.usoftchina.smartschool" level="WARN"/>
-        <root level="WARN">
-            <appender-ref ref="CONSOLE_APPENDER"/>
-            <appender-ref ref="JSON_APPENDER"/>
-        </root>
-    </springProfile>
-
-</configuration>

applications/wechat/wechat-server/src/main/resources/mapper/BrokermessagelogMapper.xml

@@ -0,0 +1,128 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.usoftchina.smartschool.wechat.mapper.BrokerMessageLogMapper" >
+  <resultMap id="BaseResultMap" type="com.usoftchina.smartschool.wechat.po.BrokerMessagelog" >
+    <id column="id" property="id" jdbcType="INTEGER" />
+    <result column="msgId" property="msgId" jdbcType="VARCHAR" />
+    <result column="message" property="message" jdbcType="LONGVARCHAR" />
+    <result column="status" property="status" jdbcType="VARCHAR" />
+    <result column="createTime" property="createTime" jdbcType="TIMESTAMP" />
+    <result column="reason" property="reason" jdbcType="VARCHAR" />
+    <result column="retry" property="retry" jdbcType="INTEGER" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, msgId, message, status, createTime, reason, retry
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.Integer" >
+    select 
+    <include refid="Base_Column_List" />
+    from brokermessagelog
+    where id = #{id,jdbcType=INTEGER}
+  </select>
+
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.Integer" >
+    delete from brokermessagelog
+    where id = #{id,jdbcType=INTEGER}
+  </delete>
+  <insert id="insert" parameterType="com.usoftchina.smartschool.wechat.po.BrokerMessagelog" >
+    insert into brokermessagelog (msgId, message,
+      status, createTime, reason
+      )
+    values ( #{msgId,jdbcType=VARCHAR}, #{message,jdbcType=VARCHAR},
+      #{status,jdbcType=VARCHAR}, #{createTime,jdbcType=TIMESTAMP}, #{reason,jdbcType=VARCHAR}
+      )
+  </insert>
+  <insert id="insertSelective" parameterType="com.usoftchina.smartschool.wechat.po.BrokerMessagelog" >
+    insert into brokermessagelog
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="msgId != null" >
+        msgId,
+      </if>
+      <if test="message != null" >
+        message,
+      </if>
+      <if test="status != null" >
+        status,
+      </if>
+      <if test="createTime != null" >
+        createTime,
+      </if>
+      <if test="reason != null" >
+        reason,
+      </if>
+      <if test="retry != null" >
+        retry,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="msgId != null" >
+        #{msgId,jdbcType=VARCHAR},
+      </if>
+      <if test="message != null" >
+        #{message,jdbcType=LONGVARCHAR},
+      </if>
+      <if test="status != null" >
+        #{status,jdbcType=VARCHAR},
+      </if>
+      <if test="createTime != null" >
+        #{createTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="reason != null" >
+        #{reason,jdbcType=VARCHAR},
+      </if>
+      <if test="retry != null" >
+        #{retry},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.usoftchina.smartschool.wechat.po.BrokerMessagelog" >
+    update brokermessagelog
+    <set >
+      <if test="message != null" >
+        message = #{message,jdbcType=VARCHAR},
+      </if>
+      <if test="status != null" >
+        status = #{status,jdbcType=VARCHAR},
+      </if>
+      <if test="createTime != null" >
+        createTime = #{createTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="reason != null" >
+        reason = #{reason,jdbcType=VARCHAR},
+      </if>
+    </set>
+    where msgId = #{msgId,jdbcType=VARCHAR}
+  </update>
+
+  <update id="updateMessageLogandAddRetry" parameterType="com.usoftchina.smartschool.wechat.po.BrokerMessagelog" >
+    update brokermessagelog
+    set  status = #{status}, reason = #{reason}, retry = retry + 1
+    where msgId = #{msgId,jdbcType=VARCHAR}
+  </update>
+
+  <update id="updateByPrimaryKey" parameterType="com.usoftchina.smartschool.wechat.po.BrokerMessagelog" >
+    update brokermessagelog
+    set msgId = #{msgId,jdbcType=VARCHAR},
+      message = #{message,jdbcType=VARCHAR},
+      status = #{status,jdbcType=VARCHAR},
+      createTime = #{createTime,jdbcType=TIMESTAMP},
+      reason = #{reason,jdbcType=VARCHAR}
+    where id = #{id,jdbcType=INTEGER}
+  </update>
+
+  <select id="getMessage" parameterType="string" resultMap="BaseResultMap">
+    select * from BrokerMessagelog where msgId = #{msgId}
+  </select>
+
+  <select id="getLiveMessage" parameterType="string" resultMap="BaseResultMap">
+    select * from BrokerMessagelog where msgId = #{msgId} and retry &lt; 3
+  </select>
+
+  <select id="getDeadMessage" parameterType="string" resultMap="BaseResultMap">
+    select * from BrokerMessagelog where msgId = #{msgId} and retry &gt; 3
+  </select>
+
+  <update id="updateSuccess" parameterType="string">
+    update BrokerMessagelog set status='投递成功',reason='' where msgid = #{msgId}
+  </update>
+</mapper>

framework/core/src/main/java/com/usoftchina/smartschool/exception/ExceptionCode.java

@@ -63,7 +63,8 @@ public enum ExceptionCode implements BaseExceptionCode {
     TURNIN_EXIST(60000,"已入库"),
     TURNINNUM_NOT_EXIST(60001,"该采购单已全部转验收,无法转采购验收单"),
 
-    CLOSED_EXIST(60002,"已关闭")
+    CLOSED_EXIST(60002,"已关闭"),
+    NULL_DATA(60003,"无效数据")
 
     ;