Add RabbitMQ support and implement message synchronization for org operations:

- Added `spring-boot-starter-amqp` dependency in POM. - Introduced classes `CorrelationDataEx`, `RetryTask`, `RetryTaskHolder`, and `RabbitMessageSender` for RabbitMQ message handling and retry logic. - Updated `OrgController` to send RabbitMQ notifications for add, update, and delete operations. - Modified `bootstrap.yml` to include `rabbitmq.yml` in shared Nacos configurations.

Add RabbitMQ support and implement message synchronization for org operations:
- Added `spring-boot-starter-amqp` dependency in POM. - Introduced classes `CorrelationDataEx`, `RetryTask`, `RetryTaskHolder`, and `RabbitMessageSender` for RabbitMQ message handling and retry logic. - Updated `OrgController` to send RabbitMQ notifications for add, update, and delete operations. - Modified `bootstrap.yml` to include `rabbitmq.yml` in shared Nacos configurations.
shaofan
1 parent b62efd8c
Showing 7 changed files with 340 additions and 1 deletions
itcast-auth/itcast-auth-server/pom.xml
itcast-auth/itcast-auth-server/src/main/java/com/itheima/authority/biz/mq/CorrelationDataEx.java
itcast-auth/itcast-auth-server/src/main/java/com/itheima/authority/biz/mq/RabbitMessageSender.java
itcast-auth/itcast-auth-server/src/main/java/com/itheima/authority/biz/mq/RetryTask.java
itcast-auth/itcast-auth-server/src/main/java/com/itheima/authority/biz/mq/RetryTaskHolder.java
itcast-auth/itcast-auth-server/src/main/java/com/itheima/authority/controller/core/OrgController.java
itcast-auth/itcast-auth-server/src/main/resources/bootstrap.yml
@@ -52,6 +52,10 @@
             <artifactId>itcast-tools-core</artifactId>
         </dependency>
         <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
+        <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-openfeign</artifactId>
             <exclusions>
+package com.itheima.authority.biz.mq;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+
+/**
+ * 扩展
+ */
+public class CorrelationDataEx extends CorrelationData {
+
+    private Message message;
+    private String exchange;
+    private String routingKey;
+    private int maxRetry;
+    private int retryCount;
+
+    public CorrelationDataEx(String id) {
+        super(id);
+    }
+
+    public Message getMessage() {
+        return message;
+    }
+
+    public void setMessage(Message message) {
+        this.message = message;
+    }
+
+    public String getExchange() {
+        return exchange;
+    }
+
+    public void setExchange(String exchange) {
+        this.exchange = exchange;
+    }
+
+    public String getRoutingKey() {
+        return routingKey;
+    }
+
+    public void setRoutingKey(String routingKey) {
+        this.routingKey = routingKey;
+    }
+
+    public int getMaxRetry() {
+        return maxRetry;
+    }
+
+    public void setMaxRetry(int maxRetry) {
+        this.maxRetry = maxRetry;
+    }
+
+    public int getRetryCount() {
+        return retryCount;
+    }
+
+    public void setRetryCount(int retryCount) {
+        this.retryCount = retryCount;
+    }
+
+    public boolean hasNext() {
+        this.retryCount += 1;
+        return this.retryCount < this.maxRetry;
+    }
+}
+package com.itheima.authority.biz.mq;
+
+import cn.hutool.core.util.StrUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageBuilder;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+
+/**
+ * rabbitmq 消息发送
+ */
+@Component
+public class RabbitMessageSender {
+    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMessageSender.class);
+
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    private RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
+        @Override
+        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
+            if (!ack) {
+                if (!(correlationData instanceof CorrelationDataEx)) {
+                    return;
+                }
+                retry((CorrelationDataEx) correlationData);
+            }
+        }
+    };
+
+    private RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
+        @Override
+        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
+            LOGGER.info("消息报文:{}", new String(message.getBody()));
+            LOGGER.info("消息编号:{}", replyCode);
+            LOGGER.info("描述:{}", replyText);
+            LOGGER.info("交换机名称:{}", exchange);
+            LOGGER.info("路由名称:{}", routingKey);
+        }
+    };
+
+    private void retry(CorrelationDataEx correlationDataEx) {
+        if (correlationDataEx.hasNext()) {
+            RetryTaskHolder.offer(new RetryTask(1, 5000, 5000, () -> send(correlationDataEx)));
+        }
+    }
+
+    private boolean send(CorrelationDataEx correlationDataEx) {
+        this.rabbitTemplate.setMandatory(true);
+        this.rabbitTemplate.setConfirmCallback(this.confirmCallback);
+        this.rabbitTemplate.setReturnCallback(this.returnCallback);
+        try {
+            this.rabbitTemplate.convertAndSend(correlationDataEx.getExchange(), correlationDataEx.getRoutingKey(), correlationDataEx.getMessage(), correlationDataEx);
+            return true;
+        } catch (AmqpException e) {
+            LOGGER.error("Rabbit mq send error", e);
+            return false;
+        }
+    }
+
+    /**
+     * 发送消息
+     *
+     * @param exchange
+     * @param routingKey
+     * @param json
+     * @param maxRetry
+     */
+    public void send(String exchange, String routingKey, String json, int maxRetry) {
+        if (StrUtil.isEmpty(exchange) || StrUtil.isEmpty(json)) {
+            return;
+        }
+        String uuid = UUID.randomUUID().toString();
+        Message message = MessageBuilder.withBody(json.getBytes(StandardCharsets.UTF_8)).setContentType("application/json").setContentEncoding("UTF-8").setMessageId(uuid).build();
+        CorrelationDataEx correlationData = new CorrelationDataEx(uuid);
+        correlationData.setMessage(message);
+        correlationData.setExchange(exchange);
+        correlationData.setRoutingKey(routingKey);
+        correlationData.setMaxRetry(maxRetry);
+        if (!send(correlationData)) {
+            retry(correlationData);
+        }
+    }
+}
+package com.itheima.authority.biz.mq;
+
+
+import javax.validation.constraints.NotNull;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+
+/** 重试任务 */
+public class RetryTask implements Delayed {
+
+    /** 执行时间 毫秒 */
+    private long mills;
+    /** 时间间隔 毫秒 */
+    private long interval;
+    /** 最大执行次数 */
+    private int max;
+    /** 当前执行次数 */
+    private int current;
+    /** 执行任务 */
+    private Callable<Boolean> callable;
+
+    /**
+     * 构造函数
+     * @param max 最大执行次数
+     * @param interval 时间间隔 毫秒
+     * @param callable 任务
+     */
+    public RetryTask(int max, long interval, Callable<Boolean> callable) {
+        this.mills = System.currentTimeMillis();
+        this.max = max;
+        this.interval = interval;
+        this.callable = callable;
+    }
+
+    /**
+     * 构造函数
+     * @param max 最大执行次数
+     * @param interval 时间间隔 毫秒
+     * @param offset 时间偏移 毫秒
+     * @param callable 任务
+     */
+    public RetryTask(int max, long interval, long offset, Callable<Boolean> callable) {
+        this.mills = System.currentTimeMillis() + offset;
+        this.max = max;
+        this.interval = interval;
+        this.callable = callable;
+    }
+
+    @Override
+    public long getDelay(@NotNull TimeUnit unit) {
+        long temp = this.mills - System.currentTimeMillis();
+        return unit.convert(temp, TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public int compareTo(@NotNull Delayed delayed) {
+        RetryTask retryTask = (RetryTask)delayed;
+        return this.mills - retryTask.getMills() < 0 ? -1 : 1;
+    }
+
+    public long getMills() {
+        return mills;
+    }
+
+    public void setMills(long mills) {
+        this.mills = mills;
+    }
+
+    public long getInterval() {
+        return interval;
+    }
+
+    public void setInterval(long interval) {
+        this.interval = interval;
+    }
+
+    public int getMax() {
+        return max;
+    }
+
+    public void setMax(int max) {
+        this.max = max;
+    }
+
+    public int getCurrent() {
+        return current;
+    }
+
+    public void setCurrent(int current) {
+        this.current = current;
+    }
+
+    public Callable<Boolean> getCallable() {
+        return callable;
+    }
+
+    public void setCallable(Callable<Boolean> callable) {
+        this.callable = callable;
+    }
+
+    public boolean hasNext() {
+        this.current += 1;
+        this.mills = System.currentTimeMillis() + this.interval;
+        return this.current < this.max;
+    }
+}
+package com.itheima.authority.biz.mq;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.core.task.AsyncTaskExecutor;
+
+import java.util.concurrent.DelayQueue;
+
+/**
+ * 重试任务处理
+ */
+public class RetryTaskHolder {
+    private static final Logger LOGGER = LoggerFactory.getLogger(RetryTaskHolder.class);
+
+    private static final DelayQueue<RetryTask> RETRY_TASK_QUEUE = new DelayQueue<>();
+
+    //直接开启线程执行
+    static {
+        int availableProcessors = Runtime.getRuntime().availableProcessors();
+        for (int i = 0; i < availableProcessors; i++) {
+
+        }
+    }
+
+    /**
+     * 添加延时任务
+     * @param retryTask
+     * @return
+     */
+    public static boolean offer(RetryTask retryTask) {
+        return RETRY_TASK_QUEUE.offer(retryTask);
+    }
+}
 package com.itheima.authority.controller.core;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.google.common.collect.Lists;
+import com.itheima.authority.biz.mq.RabbitMessageSender;
 import com.itheima.authority.biz.service.auth.UserService;
 import com.itheima.authority.biz.service.core.OrgService;
 import com.itheima.authority.dto.core.OrgSaveDTO;
@@ -56,6 +60,8 @@ public class OrgController extends BaseController {
     private UserService userService;
     @Autowired
     private DozerUtils dozer;
+    @Autowired
+    private RabbitMessageSender rabbitMessageSender;
 //    /**
 //     * 分页查询组织
@@ -126,6 +132,13 @@ public class OrgController extends BaseController {
             org.setTreePath(StringUtils.join(parent.getTreePath(), parent.getId(), DEF_ROOT_PATH));
         }
         this.orgService.saveOrg(org);
+
+        // 同步mq通知修改neo4j信息
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.put("type", "ORG");
+        jsonObject.put("operation", "ADD");
+        jsonObject.put("content", Lists.newArrayList(org));
+        rabbitMessageSender.send("itcast-auth", "add", jsonObject.toString(), 3);
         return this.success(org);
     }
@@ -135,11 +148,20 @@ public class OrgController extends BaseController {
     public R<Org> update(@RequestBody @Validated(SuperEntity.Update.class) OrgUpdateDTO data) {
         Org org = this.dozer.map(data, Org.class);
         this.orgService.updateOrg(org);
+
+        // 同步mq通知修改neo4j信息
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.put("type", "ORG");
+        jsonObject.put("operation", "UPDATE");
+        jsonObject.put("content", Lists.newArrayList(org));
+        rabbitMessageSender.send("itcast-auth", "update", jsonObject.toString(), 3);
+
         if (data.getStatus() != null) {
             org = new Org();
             org.setId(data.getId());
             org.setStatus(data.getStatus());
             this.orgService.updateStatus(org);
+
         }
         return this.success(org);
     }
@@ -168,7 +190,23 @@ public class OrgController extends BaseController {
     @SysLog("删除组织")
     @DeleteMapping
     public R<List<Org>> delete(@RequestParam("ids[]") List<Long> ids) {
+        // 查询待删除的组织
+        List<Org> delList = new ArrayList<>();
+        for (Long id : ids) {
+            Org org = this.orgService.findById(id);
+            delList.add(org);
+        }
         List<Org> orgs = this.orgService.remove(ids);
+
+        // 同步mq通知修改neo4j信息
+        for (Org org : delList) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("type", "ORG");
+            jsonObject.put("operation", "DEL");
+            jsonObject.put("content", Lists.newArrayList(org));
+            rabbitMessageSender.send("itcast-auth", "DEL", jsonObject.toString(), 3);
+        }
+
         return this.success(orgs);
     }
@@ -20,7 +20,7 @@ spring:
         namespace: aa86e672-3e5a-4480-9910-dd70bb65a7bc
         #支持多个共享 Data Id 的配置，多个之间用逗号隔开,多个共享配置间的一个优先级的关系我们约定：按照配置出现的先后顺序，即后面的优先级要高于前面
         #Data Id 必须带文件扩展名，文件扩展名既可支持 properties，也可以支持 yaml/yml。 此时 spring.cloud.nacos.config.file-extension 的配置对自定义扩展配置的 Data Id 文件扩展名没有影响。
-        shared-dataids: common.yml,redis.yml,mysql.yml
+        shared-dataids: common.yml,redis.yml,mysql.yml,rabbitmq.yml
         #支持哪些共享配置的 Data Id 在配置变化时，应用中是否可动态刷新， 感知到最新的配置值，多个 Data Id 之间用逗号隔开。如果没有明确配置，默认情况下所有共享配置的 Data Id 都不支持动态刷新。‘
         refreshable-dataids: common.yml
         enabled: true