作者 王勇

删除交换机,级联删除对应的路由键和配置关系

... ... @@ -4,7 +4,7 @@ import com.sunyo.wlpt.message.bus.service.domain.XmlData;
import com.sunyo.wlpt.message.bus.service.exception.CustomExceptionType;
import com.sunyo.wlpt.message.bus.service.rabbit.utils.DirectUtils;
import com.sunyo.wlpt.message.bus.service.response.ResultJson;
import com.sunyo.wlpt.message.bus.service.service.AsyncTaskService;
import com.sunyo.wlpt.message.bus.service.service.impl.AsyncTaskService;
import com.sunyo.wlpt.message.bus.service.service.UserMessageBindingService;
import com.sunyo.wlpt.message.bus.service.utils.XmlUtils;
import lombok.extern.slf4j.Slf4j;
... ...
... ... @@ -92,4 +92,12 @@ public interface RoutingKeyMapper {
* @return List<RoutingKey>
*/
List<RoutingKey> selectRoutingKeyExist(RoutingKey routingKey);
/**
* 根据交换机id,删除路由键信息
*
* @param exchangeId 交换机id
* @return
*/
int deleteByExchangeId(String exchangeId);
}
... ...
... ... @@ -23,6 +23,22 @@ public interface UserMessageBindingMapper {
int deleteByPrimaryKey(String id);
/**
* 根据交换机id删除配置关系
*
* @param exchangeId 交换机id
* @return 删除成功
*/
int deleteByExchangeId(String exchangeId);
/**
* 根据队列Id,删除绑定关系
*
* @param queueId 队列id
* @return
*/
int deleteByQueueId(String queueId);
/**
* insert record to table
*
* @param record the record
... ... @@ -94,4 +110,5 @@ public interface UserMessageBindingMapper {
* @return 判断校验是否通过
*/
List<UserMessageBinding> validateXmlBinding(XmlData xmlData);
}
\ No newline at end of file
... ...
package com.sunyo.wlpt.message.bus.service.service;
import com.sunyo.wlpt.message.bus.service.domain.XmlData;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @author 子诚
* Description:这个Service,用来保存异步任务;
* 注意点:异步任务需要单独放在一个类中
* 时间:2020/7/30 15:59
*/
@Component
public class AsyncTaskService {
@Resource
private MessageNoteService messageNoteService;
@Async
public void saveMessage(XmlData sentData)
{
// 无论消息是否发送成功,将消息存储于数据库
messageNoteService.insertMessageSelective(sentData);
}
}
... ... @@ -21,6 +21,14 @@ public interface RoutingKeyService {
int deleteByPrimaryKey(String id);
/**
* 根据交换机id,删除路由键信息
*
* @param exchangeId 交换机id
* @return
*/
int deleteByExchangeId(String exchangeId);
/**
* 新增
*
* @param record the record
... ...
... ... @@ -22,6 +22,22 @@ public interface UserMessageBindingService {
int deleteByPrimaryKey(String id) throws IOException, TimeoutException;
/**
* 根据交换机id删除配置关系
*
* @param exchangeId 交换机id
* @return 删除成功
*/
int deleteByExchangeId(String exchangeId);
/**
* 根据队列Id,删除绑定关系
*
* @param queueId 队列id
* @return
*/
int deleteByQueueId(String queueId);
/**
* 新增
*
* @param record the record
... ... @@ -77,7 +93,7 @@ public interface UserMessageBindingService {
* @param xmlData {@link XmlData} 解析之后的数据
* @return true or false
*/
public Boolean validateXmlBinding(XmlData xmlData);
Boolean validateXmlBinding(XmlData xmlData);
}
... ...
package com.sunyo.wlpt.message.bus.service.service.impl;
import com.sunyo.wlpt.message.bus.service.domain.BusExchange;
import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
import com.sunyo.wlpt.message.bus.service.domain.XmlData;
import com.sunyo.wlpt.message.bus.service.service.MessageNoteService;
import com.sunyo.wlpt.message.bus.service.service.RoutingKeyService;
import com.sunyo.wlpt.message.bus.service.service.UserMessageBindingService;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @author 子诚
* Description:这个Service,用来保存异步任务;
* 注意点:异步任务需要单独放在一个类中
* 时间:2020/7/30 15:59
*/
@Component
public class AsyncTaskService {
@Resource
private MessageNoteService messageNoteService;
@Resource
private RoutingKeyService routingKeyService;
@Resource
private UserMessageBindingService userMessageBindingService;
/**
* 异步,无论消息是否发送成功,将消息存储于数据库
*
* @param sentData {@link XmlData}
*/
@Async
public void saveMessage(XmlData sentData)
{
// 无论消息是否发送成功,将消息存储于数据库
messageNoteService.insertMessageSelective(sentData);
}
/**
* 当删除服务器的时候
*/
/**
* 当删除虚拟主机的时候
*/
/**
* 当删除交换机的时候,
* 同时,删除对应的路由键,删除包含交换机的绑定关系
*/
@Async
void exchangeCascadeDelete(BusExchange busExchange)
{
// 删除相关路由键
routingKeyService.deleteByExchangeId(busExchange.getId());
// 删除相关配置关系
userMessageBindingService.deleteByExchangeId(busExchange.getId());
}
/**
* 当删除队列的时候,删除包含队列的绑定关系
*/
@Async
void queueCascadeDelete(BusQueue busQueue)
{
// 删除相关配置关系
userMessageBindingService.deleteByQueueId(busQueue.getId());
}
}
... ...
... ... @@ -27,6 +27,9 @@ public class BusExchangeServiceImpl implements BusExchangeService {
private BusExchangeMapper busExchangeMapper;
@Resource
private AsyncTaskService asyncTaskService;
@Resource
RabbitUtils rabbitUtils;
@Override
... ... @@ -41,8 +44,12 @@ public class BusExchangeServiceImpl implements BusExchangeService {
String[] split = id.split(splitItem);
for (int i = 0; i < split.length; i++) {
BusExchange busExchange = selectByPrimaryKey(split[i]);
// 删除数据库的交换机记录
int num = busExchangeMapper.deleteByPrimaryKey(split[i]);
// 删除MQ服务器上的交换机
deleteExchange(busExchange);
// 级联删除与交换机相关的路由键,配置关系
asyncTaskService.exchangeCascadeDelete(busExchange);
if (num > 0) {
index = index + num;
}
... ... @@ -54,8 +61,12 @@ public class BusExchangeServiceImpl implements BusExchangeService {
}
} else {
BusExchange busExchange = selectByPrimaryKey(id);
// 删除数据库的交换机记录
int num = busExchangeMapper.deleteByPrimaryKey(id);
// 删除MQ服务器上的交换机
deleteExchange(busExchange);
// 级联删除与交换机相关的路由键,配置关系
asyncTaskService.exchangeCascadeDelete(busExchange);
return num;
}
}
... ...
... ... @@ -31,6 +31,8 @@ public class BusQueueServiceImpl implements BusQueueService {
@Resource
private RabbitUtils rabbitUtils;
@Resource
private AsyncTaskService asyncTaskService;
@Override
@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
... ... @@ -44,8 +46,12 @@ public class BusQueueServiceImpl implements BusQueueService {
String[] split = id.split(splitItem);
for (int i = 0; i < split.length; i++) {
BusQueue busQueue = selectByPrimaryKey(split[i]);
// 删除数据库中的该队列
int num = busQueueMapper.deleteByPrimaryKey(split[i]);
// 删除MQ服务器上的队列
deleteQueue(busQueue);
// 级联删除数据库中与队列有关的配置关系
asyncTaskService.queueCascadeDelete(busQueue);
if (num > 0) {
index = index + num;
}
... ... @@ -57,8 +63,12 @@ public class BusQueueServiceImpl implements BusQueueService {
}
} else {
BusQueue busQueue = selectByPrimaryKey(id);
// 删除数据库中的该队列
int num = busQueueMapper.deleteByPrimaryKey(id);
// 删除MQ服务器上的队列
deleteQueue(busQueue);
// 级联删除数据库中与队列有关的配置关系
asyncTaskService.queueCascadeDelete(busQueue);
return num;
}
}
... ...
... ... @@ -51,6 +51,12 @@ public class RoutingKeyServiceImpl implements RoutingKeyService {
}
@Override
public int deleteByExchangeId(String exchangeId)
{
return routingKeyMapper.deleteByExchangeId(exchangeId);
}
@Override
public int insert(RoutingKey record)
{
return routingKeyMapper.insert(record);
... ...
... ... @@ -423,6 +423,18 @@ public class UserMessageBindingServiceImpl implements UserMessageBindingService
List<UserMessageBinding> list = userMessageBindingMapper.validateXmlBinding(xmlData);
return list.size() > 0;
}
@Override
public int deleteByExchangeId(String exchangeId)
{
return userMessageBindingMapper.deleteByExchangeId(exchangeId);
}
@Override
public int deleteByQueueId(String queueId)
{
return userMessageBindingMapper.deleteByQueueId(queueId);
}
}
... ...
... ... @@ -107,6 +107,13 @@
from routing_key
where id = #{id,jdbcType=VARCHAR}
</delete>
<delete id="deleteByExchangeId" parameterType="java.lang.String">
<!--@mbg.generated-->
delete
from routing_key
where exchange_id = #{exchangeId,jdbcType=VARCHAR}
</delete>
<insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.domain.RoutingKey">
<!--@mbg.generated-->
insert into routing_key (id, routing_key_name, exchange_id,
... ...
... ... @@ -33,11 +33,29 @@
from user_message_binding
where id = #{id,jdbcType=VARCHAR}
</select>
<delete id="deleteByPrimaryKey" parameterType="java.lang.String">
<!--@mbg.generated-->delete
from user_message_binding
where id = #{id,jdbcType=VARCHAR}
<!--@mbg.generated-->
delete
from user_message_binding
where id = #{id,jdbcType=VARCHAR}
</delete>
<!-- 根据交换机id,删除配置关系-->
<delete id="deleteByExchangeId" parameterType="java.lang.String">
<!--@mbg.generated-->
delete
from user_message_binding
where exchange_id = #{exchangeId,jdbcType=VARCHAR}
</delete>
<!-- 根据队列id,删除配置关系-->
<delete id="deleteByQueueId" parameterType="java.lang.String">
<!--@mbg.generated-->
delete
from user_message_binding
where queue_id = #{queueId,jdbcType=VARCHAR}
</delete>
<insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.domain.UserMessageBinding">
<!--@mbg.generated-->
insert into user_message_binding (id, user_id, username,
... ... @@ -354,7 +372,7 @@
<select id="validateXmlBinding" parameterType="com.sunyo.wlpt.message.bus.service.domain.XmlData"
resultMap="BaseResultMap">
select
select
<include refid="Base_Column_List"/>
from user_message_binding
<where>
... ...