接入消息队列

build.gradle

@@ -32,8 +32,10 @@ dependencies {
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
     implementation libs.httpclient5
-    implementation 'org.bouncycastle:bcprov-jdk18on:1.80'
-    implementation 'com.google.code.gson:gson:2.12.1'
+    implementation libs.bouncycastle
+    implementation libs.gson
+    implementation  'org.apache.pulsar:pulsar-client:4.0.4'
+    implementation 'org.springframework.pulsar:spring-pulsar:1.2.4'
 }
 
 tasks.named('test') {

gradle/libs.versions.toml

@@ -3,6 +3,8 @@
 [libraries]
 springbootstartWeb = {module="org.springframework.boot:spring-boot-starter-web",version="3.4.4"}
 httpclient5 = {module="org.apache.httpcomponents.client5:httpclient5",version="5.4.1"}
+bouncycastle= {module="org.bouncycastle:bcprov-jdk18on",version="1.80"}
+gson = {module="com.google.code.gson:gson",version="2.12.1"}
 [plugins]
 
 [bundles]

settings.gradle

@@ -2,13 +2,24 @@
 pluginManagement {
     repositories {
         //阿里仓库加速
-        maven { url 'https://maven.aliyun.com/repository/google' }
-        maven { url 'https://maven.aliyun.com/repository/gradle-plugin' }
-        maven { url 'https://maven.aliyun.com/repository/public' }
-        maven { url 'https://maven.aliyun.com/repository/jcenter' }
-        //网易仓库加速
-        maven {url 'https://mirrors.163.com/maven/repository/maven-public/'}
-        maven {url 'https://mirrors.163.com/maven/repository/maven-central/'}
+        maven {
+            url 'https://maven.aliyun.com/repository/google'
+            //如果阿里仓库找不到,就去华为加速
+            artifactUrls 'https://repo.huaweicloud.com/repository/maven'
+        }
+        maven {
+            url 'https://maven.aliyun.com/repository/gradle-plugin'
+            artifactUrls 'https://repo.huaweicloud.com/repository/maven'
+        }
+        maven {
+            url 'https://maven.aliyun.com/repository/public'
+            artifactUrls 'https://repo.huaweicloud.com/repository/maven'
+        }
+        maven {
+            url 'https://maven.aliyun.com/repository/jcenter'
+            artifactUrls 'https://repo.huaweicloud.com/repository/maven'
+        }
+
         //maven { url "https://kotlin.bintray.com/kotlinx/" }
         //原生仓库地址
         google()
@@ -24,13 +35,23 @@ dependencyResolutionManagement {
     repositoriesMode.set(RepositoriesMode.FAIL_ON_PROJECT_REPOS)
     repositories {
         //阿里仓库加速
-        maven { url 'https://maven.aliyun.com/repository/google' }
-        maven { url 'https://maven.aliyun.com/repository/gradle-plugin' }
-        maven { url 'https://maven.aliyun.com/repository/public' }
-        maven { url 'https://maven.aliyun.com/repository/jcenter' }
-        //网易仓库加速
-        maven {url 'https://mirrors.163.com/maven/repository/maven-public/'}
-        maven {url 'https://mirrors.163.com/maven/repository/maven-central/'}
+        maven {
+            url 'https://maven.aliyun.com/repository/google'
+            artifactUrls 'https://repo.huaweicloud.com/repository/maven'
+        }
+        maven {
+            url 'https://maven.aliyun.com/repository/gradle-plugin'
+            artifactUrls 'https://repo.huaweicloud.com/repository/maven'
+        }
+        maven {
+            url 'https://maven.aliyun.com/repository/public'
+            artifactUrls 'https://repo.huaweicloud.com/repository/maven'
+        }
+        maven {
+            url 'https://maven.aliyun.com/repository/jcenter'
+            artifactUrls 'https://repo.huaweicloud.com/repository/maven'
+        }
+
         //原生仓库地址
         //maven { url "https://kotlin.bintray.com/kotlinx/" }
         google()

src/main/java/cn/jx/wechat/app/wechatrobot/config/AutoBeanFilterConfig.java

@@ -16,8 +16,9 @@ public class AutoBeanFilterConfig implements AutoConfigurationImportFilter, Envi
     private Map<String, List<String>> filterConfigMap = new LinkedHashMap<>();
 
     public AutoBeanFilterConfig(){
-        //Set<String> autoHttpClient5Filter=new HashSet<>(2);
-        //this.filterConfigMap.put("auto.httpClient5",autoHttpClient5Filter);
+        List<String> autoPulsarFilter=new ArrayList<>(1);
+        autoPulsarFilter.add("org.springframework.boot.autoconfigure.pulsar.PulsarAutoConfiguration");
+        this.filterConfigMap.put("auto.Pulsar",autoPulsarFilter);
     }
 
     @Override

src/main/java/cn/jx/wechat/app/wechatrobot/config/PulsarMsgQueConfig.java

@@ -0,0 +1,46 @@
+package cn.jx.wechat.app.wechatrobot.config;
+
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.core.*;
+import org.springframework.pulsar.transaction.PulsarTransactionManager;
+
+
+@Configuration
+public class PulsarMsgQueConfig {
+
+    @Bean("systemPulsarClientFactory")
+    public DefaultPulsarClientFactory pulsarClientFactory() {
+        return new DefaultPulsarClientFactory(clientBuilder -> {
+            clientBuilder.serviceUrl("pulsar://127.0.0.1:6650");
+        });
+    }
+
+    @Bean("systemPulsarClient")
+    public PulsarClient pulsarClient(@Qualifier("systemPulsarClientFactory") PulsarClientFactory clientFactory) {
+        return clientFactory.createClient();
+    }
+
+
+
+
+//
+//    @Bean("systemPulsarProducerFactory")
+//    public DefaultPulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient) {
+//        DefaultPulsarProducerFactory<?> producerFactory = new DefaultPulsarProducerFactory(pulsarClient,"", topicResolver);
+//        return producerFactory;
+//    }
+
+
+
+
+
+    @Bean("systemPulsarTransactionManager")
+    public PulsarTransactionManager pulsarTransactionManager(@Qualifier("systemPulsarClient") PulsarClient pulsarClient) {
+        return new PulsarTransactionManager(pulsarClient);
+    }
+
+}

src/main/java/cn/jx/wechat/app/wechatrobot/service/ServiceConfig.java

@@ -1,6 +1,8 @@
 package cn.jx.wechat.app.wechatrobot.service;
 
+import cn.jx.wechat.app.wechatrobot.service.face.AppApi;
 import cn.jx.wechat.app.wechatrobot.service.face.BusinessVerify;
+import cn.jx.wechat.app.wechatrobot.service.faceInst.AppApiInst;
 import cn.jx.wechat.app.wechatrobot.service.faceInst.BusinessVerifyInst;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -11,4 +13,9 @@ public class ServiceConfig {
     public BusinessVerify registryBusinessVerify(){
         return new BusinessVerifyInst();
     }
+
+    @Bean
+    public AppApi registryAppApi(){
+        return new AppApiInst();
+    }
 }

src/main/java/cn/jx/wechat/app/wechatrobot/service/face/AppApi.java

@@ -0,0 +1,9 @@
+package cn.jx.wechat.app.wechatrobot.service.face;
+
+import cn.jx.wechat.app.wechatrobot.service.WechatFaceService;
+import com.google.gson.JsonObject;
+
+public interface AppApi extends WechatFaceService {
+
+    JsonObject sendMessage(String msg, String msgType, String toUser);
+}

src/main/java/cn/jx/wechat/app/wechatrobot/service/faceInst/AppApiInst.java

@@ -0,0 +1,114 @@
+package cn.jx.wechat.app.wechatrobot.service.faceInst;
+
+import cn.jx.wechat.app.wechatrobot.config.BusinessAccessTokenConfig;
+import cn.jx.wechat.app.wechatrobot.config.BusinessVerifyProperties;
+import cn.jx.wechat.app.wechatrobot.service.face.AppApi;
+import cn.jx.wechat.app.wechatrobot.util.SpringBeanHelper;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.*;
+import org.springframework.web.client.RestTemplate;
+
+public class AppApiInst implements AppApi {
+    private static Logger log= LoggerFactory.getLogger(AppApiInst.class);
+
+    private final static  String Send_Message_Url="https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=%s";
+    @Override
+    public JsonObject sendMessage(String msg, String msgType, String toUser) {
+        JsonObject result = new JsonObject();
+        result.addProperty("code", 0);
+        result.addProperty("mark", "success");
+        result.addProperty("msg", "");
+        result.add("data", null);
+
+        String accessToken =  null;
+        BusinessAccessTokenConfig businessAccessTokenConfig = SpringBeanHelper.getBean(BusinessAccessTokenConfig.class);
+        BusinessVerifyProperties businessVerifyProp = SpringBeanHelper.getBean(BusinessVerifyProperties.class);
+        RestTemplate rest = SpringBeanHelper.getBean("defaultAppRest", RestTemplate.class);
+        try {
+            accessToken = businessAccessTokenConfig.currentAccessToken();
+        }catch (Exception exp){
+            log.info("发送信息异常，{}", exp.getMessage());
+            result.addProperty("code", 500);
+            result.addProperty("mark", "error");
+            result.addProperty("msg", "消息发送异常:"+exp.getMessage() );
+            return result;
+        }
+        String  sendMessageUrl= String.format(Send_Message_Url, accessToken);
+
+
+
+        // 创建请求实体
+        try {
+            // 设置请求Header
+            HttpHeaders headers = new HttpHeaders();
+            headers.setContentType(MediaType.APPLICATION_JSON);
+            Gson messageGson = new GsonBuilder().setPrettyPrinting().create();
+            JsonObject messageObj = new JsonObject();
+
+            messageObj.addProperty("agentid",Integer.parseInt(businessVerifyProp.getAgentId()));
+            if(null!=msgType&&msgType.equals("text")){
+                messageObj.addProperty("msgtype","text");
+                JsonObject contentObj = new JsonObject();
+                contentObj.addProperty("content",msg);
+                messageObj.add("text",contentObj);
+            }else if(null==msgType||msgType.isEmpty()){
+                messageObj.addProperty("msgtype","text");
+                JsonObject contentObj = new JsonObject();
+                contentObj.addProperty("content",msg);
+                messageObj.add("text",contentObj);
+            }else{
+                messageObj.addProperty("msgtype",msgType);
+            }
+
+            messageObj.addProperty("touser",toUser);
+
+            HttpEntity<String> requestEntity = new HttpEntity<>(messageGson.toJson(messageObj), headers);
+            ResponseEntity<String> responseEntity = rest.exchange(sendMessageUrl, HttpMethod.POST, requestEntity, String.class);
+            if (responseEntity.getStatusCode().is2xxSuccessful()) {
+                result.addProperty("msg", "消息发送成功" );
+                String body = responseEntity.getBody();
+                if(null!=body&&!body.isEmpty()){
+                    Gson gson = new GsonBuilder().setPrettyPrinting().create();
+                    JsonObject jsonObject = gson.fromJson(responseEntity.getBody(), JsonObject.class);
+                    if(jsonObject.has("errcode")&&jsonObject.getAsInt()==60020){
+                        log.info("发送信息失败,{},{}", responseEntity.getStatusCode(), "60020");
+                        result.addProperty("msg",  "消息发送失败");
+                        result.addProperty("code", 500);
+                        result.addProperty("mark", "error");
+                        result.add("data",  jsonObject);
+                    }else{
+                        log.info("发送信息成功");
+                        result.addProperty("msg",  "发送信息成功");
+                        result.add("data",  jsonObject);
+                    }
+                }else{
+                    log.info("发送信息失败,{},{}", responseEntity.getStatusCode(), "响应为空");
+                    result.addProperty("code", 500);
+                    result.addProperty("mark", "error");
+                    result.addProperty("msg", "消息发送失败" );
+                }
+            } else {
+                log.info("发送信息失败,{},{}", responseEntity.getStatusCode(), responseEntity.getBody());
+                result.addProperty("code", 500);
+                result.addProperty("mark", "error");
+                result.addProperty("msg", "消息发送失败" );
+            }
+
+            return result;
+        }catch (Exception   exp){
+            log.info("发送信息异常，{}", exp.getMessage());
+            result.addProperty("code", 500);
+            result.addProperty("mark", "error");
+            result.addProperty("msg", "消息发送异常" );
+            return result;
+        }
+    }
+
+
+
+
+}

src/main/java/cn/jx/wechat/app/wechatrobot/web/business/AppApiCtrl.java

@@ -3,10 +3,12 @@ package cn.jx.wechat.app.wechatrobot.web.business;
 import cn.jx.wechat.app.wechatrobot.api.SendMessageBody;
 import cn.jx.wechat.app.wechatrobot.config.BusinessAccessTokenConfig;
 import cn.jx.wechat.app.wechatrobot.config.BusinessVerifyProperties;
+import cn.jx.wechat.app.wechatrobot.service.face.AppApi;
 import cn.jx.wechat.app.wechatrobot.service.face.BusinessVerify;
 import cn.jx.wechat.app.wechatrobot.util.SpringBeanHelper;
 import com.google.gson.*;
 import com.google.gson.reflect.TypeToken;
+import org.checkerframework.checker.units.qual.A;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -24,101 +26,16 @@ public class AppApiCtrl {
 
     private static Logger log= LoggerFactory.getLogger(AppApiCtrl.class);
 
-    private final static  String Send_Message_Url="https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=%s";
+    private AppApi appApi;
+    @Autowired
+    public AppApiCtrl(AppApi appApi) {
+        this.appApi = appApi;
+    }
 
     @PostMapping("/sendMassage")
     public  JsonObject sendMessage(@RequestBody SendMessageBody sendMessageBody, @RequestParam String  toUser, @RequestParam(required = false) String  msgType) {
-
-        JsonObject result = new JsonObject();
-        result.addProperty("code", 0);
-        result.addProperty("mark", "success");
-        result.addProperty("msg", "");
-        result.add("data", null);
-
-        String accessToken =  null;
-
-                BusinessAccessTokenConfig businessAccessTokenConfig = SpringBeanHelper.getBean(BusinessAccessTokenConfig.class);
-        BusinessVerifyProperties businessVerifyProp = SpringBeanHelper.getBean(BusinessVerifyProperties.class);
-        RestTemplate rest = SpringBeanHelper.getBean("defaultAppRest", RestTemplate.class);
-        try {
-            accessToken = businessAccessTokenConfig.currentAccessToken();
-        }catch (Exception exp){
-            log.info("发送信息异常，{}", exp.getMessage());
-            result.addProperty("code", 500);
-            result.addProperty("mark", "error");
-            result.addProperty("msg", "消息发送异常:"+exp.getMessage() );
-            return result;
-        }
-        String  sendMessageUrl= String.format(Send_Message_Url, accessToken);
         String  msg=  null!= sendMessageBody.getMsg()?sendMessageBody.getMsg():"";
-
-
-        // 创建请求实体
-        try {
-            // 设置请求Header
-            HttpHeaders headers = new HttpHeaders();
-            headers.setContentType(MediaType.APPLICATION_JSON);
-            Gson messageGson = new GsonBuilder().setPrettyPrinting().create();
-            JsonObject messageObj = new JsonObject();
-
-            messageObj.addProperty("agentid",Integer.parseInt(businessVerifyProp.getAgentId()));
-            if(null!=msgType&&msgType.equals("text")){
-                messageObj.addProperty("msgtype","text");
-                JsonObject contentObj = new JsonObject();
-                contentObj.addProperty("content",msg);
-                messageObj.add("text",contentObj);
-            }else if(null==msgType||msgType.isEmpty()){
-                messageObj.addProperty("msgtype","text");
-                JsonObject contentObj = new JsonObject();
-                contentObj.addProperty("content",msg);
-                messageObj.add("text",contentObj);
-            }else{
-                messageObj.addProperty("msgtype",msgType);
-            }
-
-            messageObj.addProperty("touser",toUser);
-
-            HttpEntity<String> requestEntity = new HttpEntity<>(messageGson.toJson(messageObj), headers);
-            ResponseEntity<String> responseEntity = rest.exchange(sendMessageUrl, HttpMethod.POST, requestEntity, String.class);
-            if (responseEntity.getStatusCode().is2xxSuccessful()) {
-                result.addProperty("msg", "消息发送成功" );
-                String body = responseEntity.getBody();
-                if(null!=body&&!body.isEmpty()){
-                    Gson gson = new GsonBuilder().setPrettyPrinting().create();
-                    JsonObject jsonObject = gson.fromJson(responseEntity.getBody(), JsonObject.class);
-                    if(jsonObject.has("errcode")&&jsonObject.getAsInt()==60020){
-                        log.info("发送信息失败,{},{}", responseEntity.getStatusCode(), "60020");
-                        result.addProperty("msg",  "消息发送失败");
-                        result.addProperty("code", 500);
-                        result.addProperty("mark", "error");
-                        result.add("data",  jsonObject);
-                    }else{
-                        log.info("发送信息成功");
-                        result.addProperty("msg",  "发送信息成功");
-                        result.add("data",  jsonObject);
-                    }
-                }else{
-                    log.info("发送信息失败,{},{}", responseEntity.getStatusCode(), "响应为空");
-                    result.addProperty("code", 500);
-                    result.addProperty("mark", "error");
-                    result.addProperty("msg", "消息发送失败" );
-                }
-            } else {
-                log.info("发送信息失败,{},{}", responseEntity.getStatusCode(), responseEntity.getBody());
-                result.addProperty("code", 500);
-                result.addProperty("mark", "error");
-                result.addProperty("msg", "消息发送失败" );
-            }
-
-            return result;
-        }catch (Exception   exp){
-            log.info("发送信息异常，{}", exp.getMessage());
-            result.addProperty("code", 500);
-            result.addProperty("mark", "error");
-            result.addProperty("msg", "消息发送异常" );
-            return result;
-        }
-
+        return this.appApi.sendMessage(msg,msgType,toUser);
     }
 
 

src/main/resources/application.properties

@@ -7,4 +7,8 @@ business.verify.receiveid=wwccbcc17be2b3ef86
 business.verify.token=KKAkySJjLFiKY
 business.verify.encoding-aes-key=vwN0K3aFEGb6acWlKupCWMv9BRGd9hicUcl6BLyXxtr
 business.verify.corp-secret=VjnraNtq45vKvqax68RjP8SIv2Gca7MUomzJMlehQTs
-business.verify.agent-id=1000004
+business.verify.agent-id=1000004
+
+
+
+spring.pulsar.admin.service-url=http://localhost:8080