rocketmq

hx-saas-project/pom.xml

@@ -27,7 +27,7 @@
         <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
         <spring.cloud.version>Hoxton.SR11</spring.cloud.version>
         <spring.platform.version>Cairo-SR8</spring.platform.version>
-
+        <rocketmq.version>4.9.0</rocketmq.version>
         <!-- 推荐使用Harbor -->
         <docker.registry.url>192.168.0.157</docker.registry.url>
         <docker.registry.host>http://${docker.registry.url}:2375</docker.registry.host>
@@ -110,12 +110,6 @@
             <artifactId>lombok</artifactId>
             <scope>provided</scope>
         </dependency>
-
-        <!--amqp-->
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-amqp</artifactId>
-        </dependency>
     </dependencies>
 
     <build>

hx-saas-project/saas-rocketmq/pom.xml

@@ -27,7 +27,12 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-client</artifactId>
-            <version>4.9.0</version>
+            <version>${rocketmq.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fjhx</groupId>
+            <artifactId>saas-entity</artifactId>
+            <version>${bladex.project.version}</version>
         </dependency>
     </dependencies>
 

hx-saas-project/saas-rocketmq/src/main/java/com/fjhx/rocketmq/config/RocketMQConfiguration.java

@@ -13,6 +13,7 @@ import org.springframework.context.annotation.Configuration;
 import javax.annotation.PostConstruct;
 
 /**
+ * 事务消息
  * @author caozhoujun
  * @version 1.2
  * @date 2022/7/15

hx-saas-project/saas-rocketmq/src/main/java/com/fjhx/rocketmq/config/RocketMQProperties.java

@@ -14,7 +14,7 @@ import java.util.List;
 /**
  * @author caozhoujun
  * @version 1.2
- * @date 2020/5/19
+ * @date 2022/7/15
  */
 @PropertySource("classpath:config/rocketmq.properties")
 @ConfigurationProperties(prefix = "hx.rocketmq")

hx-saas-project/saas-rocketmq/src/main/java/com/fjhx/rocketmq/config/TransactionListener.java

@@ -9,7 +9,7 @@ import java.util.concurrent.ConcurrentHashMap;
 /**
  * @author caozhoujun
  * @version 1.2
- * @date 2020/6/1
+ * @date 2022/7/15
  */
 public class TransactionListener implements org.apache.rocketmq.client.producer.TransactionListener {
 

hx-saas-project/saas-rocketmq/src/main/java/com/fjhx/rocketmq/consumer/ConsumerServiceERP.java

@@ -1,12 +1,6 @@
 package com.fjhx.rocketmq.consumer;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
-import com.quake.quake_qinxin.core.util.*;
-import com.quake.quake_qinxin.model.ESPContent;
-import com.quake.quake_qinxin.model.ESPList;
-import com.quake.quake_qinxin.model.SqlServerParam;
-import com.quake.quake_qinxin.service.IRocketmqMessageService;
-import net.sf.json.JSONArray;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
@@ -27,9 +21,6 @@ import java.util.List;
 @Component
 public class ConsumerServiceERP {
 
-    @Autowired
-    private IRocketmqMessageService service;
-
     private DefaultMQPushConsumer consumer;
 
     private String consumerGroup = "rocketmq_test_group_official_erp";
@@ -54,63 +45,6 @@ public class ConsumerServiceERP {
                         System.out.println("消费erp数据" + str);
                         JSONObject jsonObject = JSONObject.parseObject(str);
                         String whodoesp = jsonObject.get("whodoesp").toString();
-                        String receiveGroup = jsonObject.get("receiveGroup").toString();
-                        JSONArray contentArray = JSONArray.fromObject(jsonObject.get("content"));
-                        JSONArray listArray = JSONArray.fromObject(jsonObject.get("doList"));
-                        ESPContent content = new ESPContent();
-                        String uuid = UUIDUtil.getNum19();
-                        for (int i = 0; i < listArray.size(); i++) {
-                            List<List<ESPContent>> lists = StringUtil.groupList(contentArray);
-                            System.out.println(lists);
-                            ESPList espList = JSON.parseObject(listArray.get(i).toString(), ESPList.class);
-                            for(List<ESPContent> espContents:lists){
-                                System.out.println(getType(espContents));
-                                try {
-                                    WebsocketUtil.webSocketSendMessageByUsername(espList.getEspId(),StringUtil.toErsStr(whodoesp+"&"+uuid,espList, content, "start+"+receiveGroup));
-                                    //WebsocketUtil.webSocketSendMessage(jsonObject.get("group").toString(), StringUtil.toErsStr(whodoesp+"&"+uuid,espList, content, "start+"+receiveGroup));
-                                }catch (Exception e){
-                                    System.out.println("websocket服务出现错误");
-                                    EmailUtil.sendEmail("1164636026@qq.com","websocket服务出现错误,异常信息为："+e);
-                                    EmailUtil.sendEmail("504178754@qq.com","websocket服务出现错误,异常信息为："+e);
-                                }
-                                for (int j = 0; j < espContents.size(); j++) {
-                                    net.sf.json.JSONObject sfObject =  net.sf.json.JSONObject.fromObject(espContents.get(j));
-                                    ESPContent espContent = (ESPContent) net.sf.json.JSONObject.toBean(sfObject, ESPContent.class);
-                                    String param = StringUtil.toErsStr(whodoesp+"&"+uuid,espList, espContent, "");
-                                    String result = WebsocketUtil.webSocketSendMessageByUsername(espList.getEspId(),param);
-                                    //String result =  WebsocketUtil.webSocketSendMessage(jsonObject.get("group").toString(), param);
-                                    System.out.println("websocket监听发送成功与否："+result+param);
-                                    SqlServerParam p = SqlServerUtil.sqlServerParamStart(espContent,whodoesp,espList,uuid);
-                                    try {
-                                        SqlServerUtil.sqlServerSaveLog(JSON.toJSONString(p));
-                                    }catch (Exception e){
-                                        System.out.println("sqlServer服务出现错误");
-                                        EmailUtil.sendEmail("1164636026@qq.com","sqlServer服务出现错误,异常信息为："+e);
-                                        EmailUtil.sendEmail("504178754@qq.com","sqlServer服务出现错误,异常信息为："+e);
-                                    }
-                                    Thread.sleep(150);
-                                }
-                                WebsocketUtil.webSocketSendMessageByUsername(espList.getEspId(), StringUtil.toErsStr(whodoesp+"&"+uuid,espList, content, "end"));
-                                //WebsocketUtil.webSocketSendMessage(jsonObject.get("group").toString(), StringUtil.toErsStr(whodoesp+"&"+uuid,espList, content, "end"));
-                                Thread.sleep(200);
-                            }
-//                            for (int j = 0; j < contentArray.size(); j++) {
-//                                ESPContent espContent = JSON.parseObject(contentArray.get(j).toString(), ESPContent.class);
-//                                String param = StringUtil.toErsStr(whodoesp+"&"+uuid,espList, espContent, "");
-//                                String result =  WebsocketUtil.webSocketSendMessage(jsonObject.get("group").toString(), param);
-//                                System.out.println("websocket监听发送成功与否："+result);
-//                                SqlServerParam p = SqlServerUtil.sqlServerParamStart(espContent,whodoesp,espList,uuid);
-//                                try {
-//                                    SqlServerUtil.sqlServerSaveLog(JSON.toJSONString(p));
-//                                }catch (Exception e){
-//                                    System.out.println("sqlServer服务出现错误");
-//                                    EmailUtil.sendEmail("1164636026@qq.com","sqlServer服务出现错误,异常信息为："+e);
-//                                    EmailUtil.sendEmail("504178754@qq.com","sqlServer服务出现错误,异常信息为："+e);
-//                                }
-//                                Thread.sleep(150);
-//                            }
-//                            WebsocketUtil.webSocketSendMessage(jsonObject.get("group").toString(), StringUtil.toErsStr(whodoesp+"&"+uuid,espList, content, "end"));
-                        }
                         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                     } catch (Exception e) {
                         e.printStackTrace();
@@ -120,20 +54,4 @@ public class ConsumerServiceERP {
                 });
         consumer.start();
     }
-
-    /**
-     * 获取数据类型
-     * @param object
-     * @return
-     */
-    public static String getType(Object object){
-        String typeName=object.getClass().getName();
-        int length= typeName.lastIndexOf(".");
-        String type =typeName.substring(length+1);
-        return type;
-    }
-    public static void main(String[] args) {
-        int a = 0;
-        System.out.println(getType(a));
-    }
 }

hx-saas-project/saas-rocketmq/src/main/java/com/fjhx/rocketmq/producer/ProducerController.java

@@ -2,17 +2,15 @@ package com.fjhx.rocketmq.producer;
 
 import com.alibaba.fastjson.JSONObject;
 import com.fjhx.rocketmq.config.RocketMQProperties;
-import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.*;
 import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.message.MessageQueue;
+import org.springblade.core.tool.api.R;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
 
-import java.util.List;
-import java.util.concurrent.*;
+import java.util.UUID;
 
 /**
  * @author caozhoujun
@@ -30,154 +28,71 @@ public class ProducerController {
     private static final String TOPIC="rocketmq_test";
     private static final String TAG="rocketmq_official";
 
-//    /**
-//     * 同步发送消息
-//     */
-//    @PostMapping("/sendSyncMessage")
-//    private  HttpResult sync(@RequestBody JSONObject map) throws Exception {
-//        if(!map.get("topic").toString().contains("rocketmq_erp")){
-//            if(!map.get("msg").toString().contains("table")||map.get("msg").toString()==null||"".equals(map.get("msg").toString())){
-//                return HttpResult.ok(1401,"消息格式有误","fail");
-//            }
-//        }
-//        try {
-//            //创建消息
-//            Message msg = new Message();
-//            msg.setTopic(map.get("topic").toString());
-//            msg.setTags(map.get("tag").toString());
-//            msg.setKeys(String.valueOf(UUIDUtil.getUUID()));
-//            msg.setBody(JSONObject.toJSONBytes(map.get("msg")));
-//            if(map.containsKey("level")){
-//                msg.setDelayTimeLevel(Integer.parseInt(map.get("level").toString()));
-//            }
-//            //同步发送消息
-//            SendResult result = defaultProducer.send(msg);
-//            if(result.toString().contains("SEND_OK")){
-//                return HttpResult.ok(1500,"success","发送成功",result);
-//            }
-//        }catch (Exception e){
-//            EmailUtil.sendEmail("1164636026@qq.com","rq服务出现错误,异常信息为："+e);
-//            EmailUtil.sendEmail("504178754@qq.com","rq服务出现错误,异常信息为："+e);
-//            e.printStackTrace();
-//            Thread.sleep(1000);
-//        }
-//      return HttpResult.error(1401,"fail","发送失败");
-//    }
-//    /**
-//     * 异步发送消息
-//     */
-//
-//    @PostMapping("/sendAsyncMessage")
-//    private  HttpResult async(@RequestBody JSONObject map) throws Exception {
-//        if(!map.get("topic").toString().contains("rocketmq_erp")){
-//            if(!map.get("msg").toString().contains("table")||map.get("msg").toString()==null||"".equals(map.get("msg").toString())){
-//                return HttpResult.ok(1401,"消息格式有误","fail");
-//            }
-//        }
-//        //创建消息
-//        Message message = new Message(map.get("topic").toString(),map.get("tag").toString(),String.valueOf(UUIDUtil.getUUID()),JSONObject.toJSONBytes(map.get("msg")));
-//        //异步发送消息
-//        defaultProducer.send(message, new SendCallback() {
-//            @Override
-//            public void onSuccess(SendResult sendResult) {
-//                System.out.println("异步发送消息："+sendResult);
-//            }
-//            @Override
-//            public void onException(Throwable e) {
-//                e.printStackTrace();
-//                //补偿机制，根据业务情况进行使用，看是否进行重试
-//                try {
-//                    defaultProducer.setRetryTimesWhenSendFailed(5);
-//                    defaultProducer.send(message,5000L);
-//                } catch (Exception ex) {
-//                    ex.printStackTrace();
-//                }
-//            }
-//        });
-//        return HttpResult.ok(1500,"success","成功");
-//    }
-//    /**
-//     * 单项发送消息
-//     */
-//
-//    @PostMapping("/sendOneWayMessage")
-//    private  HttpResult oneWay(@RequestBody JSONObject map) throws Exception {
-//        if(!map.get("topic").toString().contains("rocketmq_erp")){
-//            if(!map.get("msg").toString().contains("table")||map.get("msg").toString()==null||"".equals(map.get("msg").toString())){
-//                return HttpResult.ok(1401,"消息格式有误","fail");
-//            }
-//        }
-//        //创建消息
-//        Message message = new Message(map.get("topic").toString(),map.get("tag").toString(),String.valueOf(UUIDUtil.getUUID()),JSONObject.toJSONBytes(map.get("msg")));
-//        //同步发送消息
-//        defaultProducer.sendOneway(message);
-//        return HttpResult.ok(1500,"success","成功");
-//    }
-//    /**
-//     * 发送事务消息，暂时没有优化
-//     * @return
-//     */
-//
-//    @PostMapping("/sendTransactionMess")
-//    public HttpResult sendTransactionMsg(@RequestBody JSONObject map) throws MQClientException {
-//        if(!map.get("topic").toString().contains("rocketmq_erp")){
-//            if(!map.get("msg").toString().contains("table")||map.get("msg").toString()==null||"".equals(map.get("msg").toString())){
-//                return HttpResult.ok(1401,"消息格式有误","fail");
-//            }
-//        }
-//        TransactionMQProducer producer = new TransactionMQProducer(rocketMQProperties.getTransactionProducerGroupName());
-//        producer.setNamesrvAddr(rocketMQProperties.getNamesrvAddr());
-//        TransactionListener transactionListener = new com.quake.quake_qinxin.rocketmq.config.TransactionListener();
-//        producer.setTransactionListener(transactionListener);
-//        producer.start();
-//        ExecutorService executorService = new ThreadPoolExecutor(
-//                2,
-//                5,
-//                100,
-//                TimeUnit.SECONDS,
-//                new ArrayBlockingQueue<Runnable>(
-//                        2000),
-//                        new ThreadFactory(){
-//                            @Override
-//                            public Thread newThread(Runnable runnable){
-//                                Thread thread = new Thread();
-//                                thread.setName("cliend-transaction-msg-check-thread");
-//                                return thread;
-//                            }
-//                        }
-//                );
-//        producer.setExecutorService(executorService);
-//        Message message = new Message(map.get("tag").toString(),map.get("topic").toString(),String.valueOf(UUIDUtil.getUUID()),JSONObject.toJSONBytes(map.get("msg")));
-//        SendResult result = producer.sendMessageInTransaction(message,"hello!");
-//        producer.shutdown();
-//        return HttpResult.ok(1500,"success","成功",result);
-//    }
-//
-//    /**
-//     * 支持顺序发送消息
-//     */
-//
-//    @PostMapping("/sendMessOrder")
-//    public HttpResult sendMsgOrder(@RequestBody JSONObject map) {
-//        SendResult result=null;
-//        if(!map.get("topic").toString().contains("rocketmq_erp")){
-//            if(!map.get("msg").toString().contains("table")||map.get("msg").toString()==null||"".equals(map.get("msg").toString())){
-//                return HttpResult.ok(1401,"消息格式有误","fail");
-//            }
-//        }
-//        Message message = new Message( map.get("topic").toString(),map.get("tag").toString(),String.valueOf(UUIDUtil.getUUID()), JSONObject.toJSONBytes(map.get("msg")));
-//        try{
-//             result = defaultProducer.send(message, new MessageQueueSelector() {
-//                @Override
-//                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
-//                    int index = ((Integer) arg) % mqs.size();
-//                    return mqs.get(index);
-//                }
-//            },1);
-//        }
-//        catch (Exception e){
-//            e.printStackTrace();
-//        }
-//        return HttpResult.ok(1500,"success","成功",result);
-//    }
+    /**
+     * 同步发送消息
+     */
+    @PostMapping("/sendSyncMessage")
+    private R sync(@RequestBody JSONObject map) throws Exception {
+        Message msg = new Message();
+        msg.setTopic(map.get("topic").toString());
+        msg.setTags(map.get("tag").toString());
+        msg.setKeys(String.valueOf(UUID.randomUUID()));
+        msg.setBody(JSONObject.toJSONBytes(map.get("msg")));
+        //同步发送消息
+        SendResult result = defaultProducer.send(msg);
+        if(result.toString().contains("SEND_OK")){
+            return R.success();
+        }
+        return R.fail("发送失败");
+    }
+    /**
+     * 异步发送消息
+     */
+    @PostMapping("/sendAsyncMessage")
+    private  R async(@RequestBody JSONObject map) throws Exception {
+        if(!map.get("topic").toString().contains("rocketmq_erp")){
+            if(!map.get("msg").toString().contains("table")||map.get("msg").toString()==null||"".equals(map.get("msg").toString())){
+                return R.fail("消息格式有误");
+            }
+        }
+        //创建消息
+        Message message = new Message(map.get("topic").toString(),map.get("tag").toString(),String.valueOf(UUID.randomUUID()),JSONObject.toJSONBytes(map.get("msg")));
+        //异步发送消息
+        defaultProducer.send(message, new SendCallback() {
+            @Override
+            public void onSuccess(SendResult sendResult) {
+                System.out.println("异步发送消息："+sendResult);
+            }
+            @Override
+            public void onException(Throwable e) {
+                e.printStackTrace();
+                //补偿机制，根据业务情况进行使用，看是否进行重试
+                try {
+                    defaultProducer.setRetryTimesWhenSendFailed(5);
+                    defaultProducer.send(message,5000L);
+                } catch (Exception ex) {
+                    ex.printStackTrace();
+                }
+            }
+        });
+        return R.success();
+    }
+    /**
+     * 单项发送消息
+     */
+
+    @PostMapping("/sendOneWayMessage")
+    private  R oneWay(@RequestBody JSONObject map) throws Exception {
+        if(!map.get("topic").toString().contains("rocketmq_erp")){
+            if(!map.get("msg").toString().contains("table")||map.get("msg").toString()==null||"".equals(map.get("msg").toString())){
+                return R.fail("消息格式有误");
+            }
+        }
+        //创建消息
+        Message message = new Message(map.get("topic").toString(),map.get("tag").toString(),String.valueOf(UUID.randomUUID()),JSONObject.toJSONBytes(map.get("msg")));
+        //同步发送消息
+        defaultProducer.sendOneway(message);
+        return R.success();
+    }
+
 }

hx-saas-project/saas-rocketmq/src/main/resources/config/rocketmq.properties

@@ -1,5 +1,5 @@
 
-hx.rocketmq.namesrvAddr=124.70.202.81:9876;124.70.177.208:9876;124.70.207.0:9876;124.70.188.19:9876
+hx.rocketmq.namesrvAddr=127.0.0.1:9876
 
 hx.rocketmq.producerGroupName=user_group_p