message_bus / message_bus_service · 提交

转到一个项目

作者 王勇
提交 d7bac85406924787465e641abbccab1f3c85aaf0 1 个父辈 6ec733a4

修改RabbitUtils中方法的返回值

内嵌 并排对比
正在显示 17 个修改的文件 包含 309 行增加 和 270 行删除
@@ -62,9 +62,7 @@ public class BusExchangeController { @@ -62,9 +62,7 @@ public class BusExchangeController {
62 public ResultJson deleteBusExchange(@RequestBody BusExchange busExchange) 62 public ResultJson deleteBusExchange(@RequestBody BusExchange busExchange)
63 { 63 {
64 try { 64 try {
65 - return busExchangeService.deleteByPrimaryKey(busExchange.getId()) > 0  
66 - ? new ResultJson<>("200", "删除MQ交换机，成功")  
67 - : new ResultJson<>("500", "删除MQ交换机，失败"); 65 + return busExchangeService.deleteByPrimaryKey(busExchange.getId());
68 } catch (IOException | TimeoutException e) { 66 } catch (IOException | TimeoutException e) {
69 return new ResultJson<>("500", "服务器异常，删除MQ交换机，失败"); 67 return new ResultJson<>("500", "服务器异常，删除MQ交换机，失败");
70 } 68 }
@@ -80,9 +78,7 @@ public class BusExchangeController { @@ -80,9 +78,7 @@ public class BusExchangeController {
80 public ResultJson batchRemoveBusExchange(String ids) 78 public ResultJson batchRemoveBusExchange(String ids)
81 { 79 {
82 try { 80 try {
83 - return busExchangeService.deleteByPrimaryKey(ids) > 0  
84 - ? new ResultJson<>("200", "批量删除MQ交换机，成功")  
85 - : new ResultJson<>("500", "批量删除MQ交换机，失败"); 81 + return busExchangeService.deleteByPrimaryKey(ids);
86 } catch (IOException | TimeoutException e) { 82 } catch (IOException | TimeoutException e) {
87 return new ResultJson<>("500", "服务器异常，批量删除MQ交换机，失败"); 83 return new ResultJson<>("500", "服务器异常，批量删除MQ交换机，失败");
88 } 84 }
@@ -122,9 +118,7 @@ public class BusExchangeController { @@ -122,9 +118,7 @@ public class BusExchangeController {
122 // 设置id 118 // 设置id
123 busExchange.setId(IdUtils.generateId()); 119 busExchange.setId(IdUtils.generateId());
124 return message == null 120 return message == null
125 - ? busExchangeService.insertSelective(busExchange) > 0  
126 - ? new ResultJson<>("200", "新增MQ交换机信息，成功")  
127 - : new ResultJson<>("500", "新增MQ交换机信息，失败") 121 + ? busExchangeService.insertSelective(busExchange)
128 : new ResultJson<>("400", message); 122 : new ResultJson<>("400", message);
129 } catch (IOException | TimeoutException e) { 123 } catch (IOException | TimeoutException e) {
130 return new ResultJson<>("500", "新增MQ交换机信息，失败"); 124 return new ResultJson<>("500", "新增MQ交换机信息，失败");
@@ -73,7 +73,7 @@ public class BusQueueController { @@ -73,7 +73,7 @@ public class BusQueueController {
73 return busQueueService.deleteByPrimaryKey(busQueue.getId()) > 0 73 return busQueueService.deleteByPrimaryKey(busQueue.getId()) > 0
74 ? new ResultJson<>("200", "删除消息队列，成功") 74 ? new ResultJson<>("200", "删除消息队列，成功")
75 : new ResultJson<>("500", "删除消息队列，失败"); 75 : new ResultJson<>("500", "删除消息队列，失败");
76 - } catch (IOException | TimeoutException e) { 76 + } catch (IOException | TimeoutException | InterruptedException e) {
77 return new ResultJson<>("500", "服务器异常，删除消息队列，失败"); 77 return new ResultJson<>("500", "服务器异常，删除消息队列，失败");
78 } 78 }
79 } 79 }
@@ -91,7 +91,7 @@ public class BusQueueController { @@ -91,7 +91,7 @@ public class BusQueueController {
91 return busQueueService.deleteByPrimaryKey(ids) > 0 91 return busQueueService.deleteByPrimaryKey(ids) > 0
92 ? new ResultJson<>("200", "批量删除MQ消息队列，成功") 92 ? new ResultJson<>("200", "批量删除MQ消息队列，成功")
93 : new ResultJson<>("500", "批量删除MQ消息队列，失败"); 93 : new ResultJson<>("500", "批量删除MQ消息队列，失败");
94 - } catch (IOException | TimeoutException e) { 94 + } catch (IOException | TimeoutException | InterruptedException e) {
95 return new ResultJson<>("500", "服务器异常，批量删除MQ消息队列，失败"); 95 return new ResultJson<>("500", "服务器异常，批量删除MQ消息队列，失败");
96 } 96 }
97 } 97 }
@@ -4,8 +4,8 @@ import com.sunyo.wlpt.message.bus.service.domain.XmlData; @@ -4,8 +4,8 @@ import com.sunyo.wlpt.message.bus.service.domain.XmlData;
4 import com.sunyo.wlpt.message.bus.service.exception.CustomExceptionType; 4 import com.sunyo.wlpt.message.bus.service.exception.CustomExceptionType;
5 import com.sunyo.wlpt.message.bus.service.rabbit.utils.DirectUtils; 5 import com.sunyo.wlpt.message.bus.service.rabbit.utils.DirectUtils;
6 import com.sunyo.wlpt.message.bus.service.response.ResultJson; 6 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
7 -import com.sunyo.wlpt.message.bus.service.service.impl.AsyncTaskService;  
8 import com.sunyo.wlpt.message.bus.service.service.UserMessageBindingService; 7 import com.sunyo.wlpt.message.bus.service.service.UserMessageBindingService;
  8 +import com.sunyo.wlpt.message.bus.service.service.impl.AsyncTaskService;
9 import com.sunyo.wlpt.message.bus.service.utils.XmlUtils; 9 import com.sunyo.wlpt.message.bus.service.utils.XmlUtils;
10 import lombok.extern.slf4j.Slf4j; 10 import lombok.extern.slf4j.Slf4j;
11 import org.dom4j.DocumentException; 11 import org.dom4j.DocumentException;
@@ -15,6 +15,8 @@ import org.springframework.web.bind.annotation.*; @@ -15,6 +15,8 @@ import org.springframework.web.bind.annotation.*;
15 import javax.annotation.Resource; 15 import javax.annotation.Resource;
16 import java.io.IOException; 16 import java.io.IOException;
17 import java.util.Date; 17 import java.util.Date;
  18 +import java.util.concurrent.CountDownLatch;
  19 +import java.util.concurrent.Future;
18 import java.util.concurrent.TimeoutException; 20 import java.util.concurrent.TimeoutException;
19 21
20 /** 22 /**
@@ -113,17 +115,20 @@ public class RabbitController { @@ -113,17 +115,20 @@ public class RabbitController {
113 } 115 }
114 // 4、mq发送消息，数据库中保存消息并保存至ES 116 // 4、mq发送消息，数据库中保存消息并保存至ES
115 return sendAndSave(sentData); 117 return sendAndSave(sentData);
116 - } catch (IOException | TimeoutException e) { 118 + } catch (IOException | TimeoutException | InterruptedException e) {
117 return ResultJson.error(CustomExceptionType.SERVER_EXCEPTION); 119 return ResultJson.error(CustomExceptionType.SERVER_EXCEPTION);
118 } 120 }
119 } 121 }
120 122
121 - public ResultJson sendAndSave(XmlData sentData) throws IOException, TimeoutException 123 + public ResultJson sendAndSave(XmlData sentData) throws IOException, TimeoutException, InterruptedException
122 { 124 {
123 // 4、mq发送消息，数据库中保存消息 125 // 4、mq发送消息，数据库中保存消息
124 ResultJson result = directUtils.sendMessage(sentData); 126 ResultJson result = directUtils.sendMessage(sentData);
  127 +
125 // 异步，保存消息记录 128 // 异步，保存消息记录
126 - asyncTaskService.saveMessage(sentData); 129 + CountDownLatch latch = new CountDownLatch(1);
  130 + Future<String> future = asyncTaskService.saveMessage(sentData, latch);
  131 + latch.await();
127 return result; 132 return result;
128 } 133 }
129 } 134 }
@@ -80,7 +80,7 @@ public class UserInfoController { @@ -80,7 +80,7 @@ public class UserInfoController {
80 } 80 }
81 81
82 /** 82 /**
83 - * 编辑用户信息 83 + * 修改用户密码
84 * 84 *
85 * @param userInfo {@link UserInfo} 85 * @param userInfo {@link UserInfo}
86 * @return 86 * @return
@@ -103,13 +103,13 @@ public class UserInfoController { @@ -103,13 +103,13 @@ public class UserInfoController {
103 { 103 {
104 try { 104 try {
105 return userInfoService.deleteUserInfo(userInfo); 105 return userInfoService.deleteUserInfo(userInfo);
106 - } catch (IOException | URISyntaxException e) { 106 + } catch (IOException | URISyntaxException | TimeoutException | InterruptedException e) {
107 return new ResultJson<>("500", "服务器异常，删除用户失败"); 107 return new ResultJson<>("500", "服务器异常，删除用户失败");
108 } 108 }
109 } 109 }
110 110
111 /** 111 /**
112 - * 删除数据关系（MQ服务器 and 数据库） 112 + * 删除用户关系（MQ服务器 and 数据库）
113 */ 113 */
114 @DeleteMapping("/deleteRelation") 114 @DeleteMapping("/deleteRelation")
115 public ResultJson deleteUserRelation(@RequestBody UserInfo userInfo) 115 public ResultJson deleteUserRelation(@RequestBody UserInfo userInfo)
@@ -2,6 +2,7 @@ package com.sunyo.wlpt.message.bus.service.mapper; @@ -2,6 +2,7 @@ package com.sunyo.wlpt.message.bus.service.mapper;
2 2
3 import com.sunyo.wlpt.message.bus.service.domain.BusQueue; 3 import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
4 import org.apache.ibatis.annotations.Mapper; 4 import org.apache.ibatis.annotations.Mapper;
  5 +import org.apache.ibatis.annotations.Param;
5 6
6 import java.util.List; 7 import java.util.List;
7 8
@@ -107,4 +108,13 @@ public interface BusQueueMapper { @@ -107,4 +108,13 @@ public interface BusQueueMapper {
107 * @return 108 * @return
108 */ 109 */
109 List<BusQueue> selectByUsername(String username); 110 List<BusQueue> selectByUsername(String username);
  111 +
  112 + /**
  113 + * 根据用户名称和虚拟主机Id查询队列信息
  114 + *
  115 + * @param username 用户名称
  116 + * @param virtualHostId 虚拟主机Id
  117 + * @return
  118 + */
  119 + List<BusQueue> selectByUsernameAndHostId(@Param("username") String username, @Param("virtualHostId") String virtualHostId);
110 } 120 }
@@ -119,6 +119,15 @@ public interface UserInfoMapper { @@ -119,6 +119,15 @@ public interface UserInfoMapper {
119 int deleteByUsername(String username); 119 int deleteByUsername(String username);
120 120
121 /** 121 /**
  122 + * 根据用户名称和服务器名称删除用户信息
  123 + *
  124 + * @param username 用户名称
  125 + * @param serverName 服务器名称
  126 + * @return
  127 + */
  128 + int deleteByUsernameAndServerName(@Param("username") String username, @Param("serverName") String serverName);
  129 +
  130 + /**
122 * 根据服务器名称，删除用户关系 131 * 根据服务器名称，删除用户关系
123 * 132 *
124 * @param serverName 服务器名称 133 * @param serverName 服务器名称
@@ -134,4 +143,12 @@ public interface UserInfoMapper { @@ -134,4 +143,12 @@ public interface UserInfoMapper {
134 */ 143 */
135 int deleteByHostName(String hostName); 144 int deleteByHostName(String hostName);
136 145
  146 + /**
  147 + * 根据用户名称和服务器名称，查询用户信息
  148 + *
  149 + * @param username 用户名称
  150 + * @param serverName 服务器名称
  151 + * @return
  152 + */
  153 + List<UserInfo> selectByUsernameAndServerName(@Param("username") String username, @Param("serverName") String serverName);
137 } 154 }
1 package com.sunyo.wlpt.message.bus.service.rabbit.utils; 1 package com.sunyo.wlpt.message.bus.service.rabbit.utils;
2 2
  3 +import com.rabbitmq.client.AMQP;
3 import com.rabbitmq.client.Channel; 4 import com.rabbitmq.client.Channel;
4 import com.rabbitmq.client.Connection; 5 import com.rabbitmq.client.Connection;
5 import com.rabbitmq.client.ConnectionFactory; 6 import com.rabbitmq.client.ConnectionFactory;
6 import com.sunyo.wlpt.message.bus.service.domain.*; 7 import com.sunyo.wlpt.message.bus.service.domain.*;
  8 +import com.sunyo.wlpt.message.bus.service.response.ResultJson;
7 import com.sunyo.wlpt.message.bus.service.service.BusServerService; 9 import com.sunyo.wlpt.message.bus.service.service.BusServerService;
8 import com.sunyo.wlpt.message.bus.service.service.VirtualHostService; 10 import com.sunyo.wlpt.message.bus.service.service.VirtualHostService;
9 import com.sunyo.wlpt.message.bus.service.utils.EncryptionUtils; 11 import com.sunyo.wlpt.message.bus.service.utils.EncryptionUtils;
@@ -106,46 +108,67 @@ public class RabbitUtils { @@ -106,46 +108,67 @@ public class RabbitUtils {
106 /** 108 /**
107 * 添加交换机 109 * 添加交换机
108 */ 110 */
109 - public void createExchange(BusServer server, String virtualHostName, BusExchange busExchange) 111 + public ResultJson createExchange(BusServer server, String virtualHostName, BusExchange busExchange)
110 throws IOException, TimeoutException 112 throws IOException, TimeoutException
111 { 113 {
  114 + String exchangeName = busExchange.getExchangeName();
  115 +
112 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword()); 116 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword());
113 Channel channel = connection.createChannel(); 117 Channel channel = connection.createChannel();
114 - channel.exchangeDeclare(busExchange.getExchangeName(), 118 + // 添加交换机成功，declareOk.toString是 #method<exchange.declare-ok>()
  119 + AMQP.Exchange.DeclareOk declareOk = channel.exchangeDeclare(exchangeName,
115 busExchange.getExchangeType(), 120 busExchange.getExchangeType(),
116 busExchange.getDurability(), 121 busExchange.getDurability(),
117 busExchange.getAutoDelete(), 122 busExchange.getAutoDelete(),
118 busExchange.getInternal(), 123 busExchange.getInternal(),
119 null); 124 null);
120 closeConnectionAndChanel(channel, connection); 125 closeConnectionAndChanel(channel, connection);
  126 +
  127 + return declareOk.toString().contains("declare-ok")
  128 + ? ResultJson.success("添加MQ交换机，成功")
  129 + : new ResultJson<>("500", "MQ服务异常，添加MQ交换机：" + exchangeName + ",失败");
121 } 130 }
122 131
123 /** 132 /**
124 * 删除交换机 channel.exchangeDelete(exchangeName); 133 * 删除交换机 channel.exchangeDelete(exchangeName);
125 */ 134 */
126 - public void removeExchange(BusServer server, String virtualHostName, BusExchange busExchange) 135 + public ResultJson removeExchange(BusServer server, String virtualHostName, BusExchange busExchange)
127 throws IOException, TimeoutException 136 throws IOException, TimeoutException
128 { 137 {
  138 + String exchangeName = busExchange.getExchangeName();
  139 +
129 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword()); 140 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword());
130 Channel channel = connection.createChannel(); 141 Channel channel = connection.createChannel();
131 - channel.exchangeDelete(busExchange.getExchangeName()); 142 + // 删除成功的时候，deleteOk.toString是"#method<exchange.delete-ok>()"
  143 + AMQP.Exchange.DeleteOk deleteOk = channel.exchangeDelete(exchangeName);
132 closeConnectionAndChanel(channel, connection); 144 closeConnectionAndChanel(channel, connection);
  145 +
  146 + return deleteOk.toString().contains("delete-ok")
  147 + ? ResultJson.success("删除MQ交换机，成功")
  148 + : new ResultJson<>("500", "MQ服务异常，删除MQ交换机：" + exchangeName + ",失败");
133 } 149 }
134 150
135 /** 151 /**
136 * 添加队列（默认设置参数为 null） 152 * 添加队列（默认设置参数为 null）
137 */ 153 */
138 - public void createQueue(BusServer server, String virtualHostName, BusQueue busQueue) 154 + public ResultJson createQueue(BusServer server, String virtualHostName, BusQueue busQueue)
139 throws IOException, TimeoutException 155 throws IOException, TimeoutException
140 { 156 {
  157 + String queueName = busQueue.getQueueName();
  158 +
141 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword()); 159 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword());
142 Channel channel = connection.createChannel(); 160 Channel channel = connection.createChannel();
143 - channel.queueDeclare(busQueue.getQueueName(), 161 + // 成功时,deleteOk.toString是"#method<exchange.declare-ok>()"
  162 + AMQP.Queue.DeclareOk declareOk = channel.queueDeclare(queueName,
144 busQueue.getDurability(), 163 busQueue.getDurability(),
145 false, 164 false,
146 busQueue.getAutoDelete(), 165 busQueue.getAutoDelete(),
147 null); 166 null);
148 closeConnectionAndChanel(channel, connection); 167 closeConnectionAndChanel(channel, connection);
  168 +
  169 + return declareOk.toString().contains("declare-ok")
  170 + ? ResultJson.success("添加MQ队列，成功")
  171 + : new ResultJson<>("500", "MQ服务异常，添加MQ队列：" + queueName + ",失败");
149 } 172 }
150 173
151 /** 174 /**
@@ -155,107 +178,113 @@ public class RabbitUtils { @@ -155,107 +178,113 @@ public class RabbitUtils {
155 /** 178 /**
156 * 删除队列 channel.queueDelete(queueName); 179 * 删除队列 channel.queueDelete(queueName);
157 */ 180 */
158 - public void removeQueue(BusServer server, String virtualHostName, BusQueue busQueue) 181 + public ResultJson removeQueue(BusServer server, String virtualHostName, BusQueue busQueue)
159 throws IOException, TimeoutException 182 throws IOException, TimeoutException
160 { 183 {
  184 + String queueName = busQueue.getQueueName();
  185 +
161 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword()); 186 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword());
162 Channel channel = connection.createChannel(); 187 Channel channel = connection.createChannel();
163 - channel.queueDelete(busQueue.getQueueName()); 188 + AMQP.Queue.DeleteOk deleteOk = channel.queueDelete(queueName);
164 closeConnectionAndChanel(channel, connection); 189 closeConnectionAndChanel(channel, connection);
  190 +
  191 + return deleteOk.toString().contains("delete-ok")
  192 + ? ResultJson.success("删除MQ队列，成功")
  193 + : new ResultJson<>("500", "MQ服务异常，删除MQ队列：" + queueName + ",失败");
165 } 194 }
166 195
167 /** 196 /**
168 * 创建绑定 197 * 创建绑定
169 */ 198 */
170 - public void createBinding(BusServer server, String virtualHostName, UserMessageBinding userMessageBinding) 199 + public ResultJson createBinding(BusServer server, String virtualHostName, UserMessageBinding userMessageBinding)
171 throws IOException, TimeoutException 200 throws IOException, TimeoutException
172 { 201 {
173 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword()); 202 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword());
174 Channel channel = connection.createChannel(); 203 Channel channel = connection.createChannel();
175 - channel.queueBind(userMessageBinding.getQueueName(),  
176 - userMessageBinding.getExchangeName(),  
177 - userMessageBinding.getRoutingKeyName()); 204 + // #method<queue.bind-ok>()
  205 + AMQP.Queue.BindOk bindOk = channel.queueBind(userMessageBinding.getQueueName(), userMessageBinding.getExchangeName(), userMessageBinding.getRoutingKeyName());
178 closeConnectionAndChanel(channel, connection); 206 closeConnectionAndChanel(channel, connection);
  207 +
  208 + return bindOk.toString().contains("bind-ok")
  209 + ? ResultJson.success("创建MQ绑定关系，成功")
  210 + : new ResultJson<>("500", "MQ服务异常，创建MQ绑定关系" + ",失败");
179 } 211 }
180 212
181 /** 213 /**
182 * 解除绑定 channel.queueUnbind("queueName", "exchangeName","routingKey"); 214 * 解除绑定 channel.queueUnbind("queueName", "exchangeName","routingKey");
183 */ 215 */
184 - public void removeBinding(BusServer server, String virtualHostName, UserMessageBinding userMessageBinding)  
185 - throws IOException, TimeoutException 216 + public ResultJson removeBinding(BusServer server, String virtualHostName, UserMessageBinding userMessageBinding) throws IOException, TimeoutException
186 { 217 {
187 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword()); 218 Connection connection = getConnection(server.getServerIp(), server.getServerPort(), virtualHostName, server.getSuperUsername(), server.getSuperPassword());
188 Channel channel = connection.createChannel(); 219 Channel channel = connection.createChannel();
189 - channel.queueUnbind(userMessageBinding.getQueueName(),  
190 - userMessageBinding.getExchangeName(),  
191 - userMessageBinding.getRoutingKeyName()); 220 + // #method<queue.unbind-ok>()
  221 + AMQP.Queue.UnbindOk unbindOk = channel.queueUnbind(userMessageBinding.getQueueName(), userMessageBinding.getExchangeName(), userMessageBinding.getRoutingKeyName());
192 closeConnectionAndChanel(channel, connection); 222 closeConnectionAndChanel(channel, connection);
  223 +
  224 + return unbindOk.toString().contains("unbind-ok")
  225 + ? ResultJson.success("解除MQ绑定关系，成功")
  226 + : new ResultJson<>("500", "MQ服务异常，解除MQ绑定关系" + ",失败");
193 } 227 }
194 228
195 /** 229 /**
196 * 前往创建交换机的路上 230 * 前往创建交换机的路上
197 */ 231 */
198 - public void toCreateExchange(BusExchange busExchange)  
199 - throws IOException, TimeoutException 232 + public ResultJson toCreateExchange(BusExchange busExchange) throws IOException, TimeoutException
200 { 233 {
201 VirtualHost virtualHost = getVirtualHost(busExchange.getVirtualHostId()); 234 VirtualHost virtualHost = getVirtualHost(busExchange.getVirtualHostId());
202 BusServer busServer = getBusServer(virtualHost.getServerId()); 235 BusServer busServer = getBusServer(virtualHost.getServerId());
203 - createExchange(busServer, virtualHost.getVirtualHostName(), busExchange); 236 + return createExchange(busServer, virtualHost.getVirtualHostName(), busExchange);
204 } 237 }
205 238
206 /** 239 /**
207 * 前往删除交换机的路上 240 * 前往删除交换机的路上
208 */ 241 */
209 - public void toRemoveExchange(BusExchange busExchange)  
210 - throws IOException, TimeoutException 242 + public ResultJson toRemoveExchange(BusExchange busExchange) throws IOException, TimeoutException
211 { 243 {
212 VirtualHost virtualHost = getVirtualHost(busExchange.getVirtualHostId()); 244 VirtualHost virtualHost = getVirtualHost(busExchange.getVirtualHostId());
213 BusServer busServer = getBusServer(virtualHost.getServerId()); 245 BusServer busServer = getBusServer(virtualHost.getServerId());
214 - removeExchange(busServer, virtualHost.getVirtualHostName(), busExchange); 246 + return removeExchange(busServer, virtualHost.getVirtualHostName(), busExchange);
215 } 247 }
216 248
217 /** 249 /**
218 * 前往创建队列的路上 250 * 前往创建队列的路上
219 */ 251 */
220 - public void toCreateQueue(BusQueue BusQueue)  
221 - throws IOException, TimeoutException 252 + public ResultJson toCreateQueue(BusQueue BusQueue) throws IOException, TimeoutException
222 { 253 {
223 VirtualHost virtualHost = getVirtualHost(BusQueue.getVirtualHostId()); 254 VirtualHost virtualHost = getVirtualHost(BusQueue.getVirtualHostId());
224 BusServer busServer = getBusServer(virtualHost.getServerId()); 255 BusServer busServer = getBusServer(virtualHost.getServerId());
225 - createQueue(busServer, virtualHost.getVirtualHostName(), BusQueue); 256 + return createQueue(busServer, virtualHost.getVirtualHostName(), BusQueue);
226 } 257 }
227 258
228 /** 259 /**
229 * 前往删除队列的路上 260 * 前往删除队列的路上
230 */ 261 */
231 - public void toRemoveQueue(BusQueue BusQueue)  
232 - throws IOException, TimeoutException 262 + public ResultJson toRemoveQueue(BusQueue BusQueue) throws IOException, TimeoutException
233 { 263 {
234 VirtualHost virtualHost = getVirtualHost(BusQueue.getVirtualHostId()); 264 VirtualHost virtualHost = getVirtualHost(BusQueue.getVirtualHostId());
235 BusServer busServer = getBusServer(virtualHost.getServerId()); 265 BusServer busServer = getBusServer(virtualHost.getServerId());
236 - removeQueue(busServer, virtualHost.getVirtualHostName(), BusQueue); 266 + return removeQueue(busServer, virtualHost.getVirtualHostName(), BusQueue);
237 } 267 }
238 268
239 /** 269 /**
240 * 前往创建绑定的路上 270 * 前往创建绑定的路上
241 */ 271 */
242 - public void toCreateBinding(UserMessageBinding userMessageBinding) 272 + public ResultJson toCreateBinding(UserMessageBinding userMessageBinding)
243 throws IOException, TimeoutException 273 throws IOException, TimeoutException
244 { 274 {
245 VirtualHost virtualHost = getVirtualHost(userMessageBinding.getVirtualHostId()); 275 VirtualHost virtualHost = getVirtualHost(userMessageBinding.getVirtualHostId());
246 BusServer busServer = getBusServer(virtualHost.getServerId()); 276 BusServer busServer = getBusServer(virtualHost.getServerId());
247 - createBinding(busServer, virtualHost.getVirtualHostName(), userMessageBinding); 277 + return createBinding(busServer, virtualHost.getVirtualHostName(), userMessageBinding);
248 } 278 }
249 279
250 /** 280 /**
251 * 前往解除绑定的路上 281 * 前往解除绑定的路上
252 */ 282 */
253 - public void toRemoveBinding(UserMessageBinding userMessageBinding)  
254 - throws IOException, TimeoutException 283 + public ResultJson toRemoveBinding(UserMessageBinding userMessageBinding) throws IOException, TimeoutException
255 { 284 {
256 VirtualHost virtualHost = getVirtualHost(userMessageBinding.getVirtualHostId()); 285 VirtualHost virtualHost = getVirtualHost(userMessageBinding.getVirtualHostId());
257 BusServer busServer = getBusServer(virtualHost.getServerId()); 286 BusServer busServer = getBusServer(virtualHost.getServerId());
258 - removeBinding(busServer, virtualHost.getVirtualHostName(), userMessageBinding); 287 + return removeBinding(busServer, virtualHost.getVirtualHostName(), userMessageBinding);
259 } 288 }
260 289
261 /** 290 /**
@@ -2,6 +2,7 @@ package com.sunyo.wlpt.message.bus.service.service; @@ -2,6 +2,7 @@ package com.sunyo.wlpt.message.bus.service.service;
2 2
3 import com.github.pagehelper.PageInfo; 3 import com.github.pagehelper.PageInfo;
4 import com.sunyo.wlpt.message.bus.service.domain.BusExchange; 4 import com.sunyo.wlpt.message.bus.service.domain.BusExchange;
  5 +import com.sunyo.wlpt.message.bus.service.response.ResultJson;
5 6
6 import java.io.IOException; 7 import java.io.IOException;
7 import java.util.List; 8 import java.util.List;
@@ -21,7 +22,7 @@ public interface BusExchangeService { @@ -21,7 +22,7 @@ public interface BusExchangeService {
21 * @param id primaryKey 22 * @param id primaryKey
22 * @return deleteCount 23 * @return deleteCount
23 */ 24 */
24 - int deleteByPrimaryKey(String id) throws IOException, TimeoutException; 25 + ResultJson deleteByPrimaryKey(String id) throws IOException, TimeoutException;
25 26
26 /** 27 /**
27 * 新增 28 * 新增
@@ -37,7 +38,7 @@ public interface BusExchangeService { @@ -37,7 +38,7 @@ public interface BusExchangeService {
37 * @param record the record 38 * @param record the record
38 * @return insert count 39 * @return insert count
39 */ 40 */
40 - int insertSelective(BusExchange record) throws IOException, TimeoutException; 41 + ResultJson insertSelective(BusExchange record) throws IOException, TimeoutException;
41 42
42 /** 43 /**
43 * 查询，根据主键 44 * 查询，根据主键
@@ -21,7 +21,7 @@ public interface BusQueueService { @@ -21,7 +21,7 @@ public interface BusQueueService {
21 * @param id primaryKey 21 * @param id primaryKey
22 * @return deleteCount 22 * @return deleteCount
23 */ 23 */
24 - int deleteByPrimaryKey(String id) throws IOException, TimeoutException; 24 + int deleteByPrimaryKey(String id) throws IOException, TimeoutException, InterruptedException;
25 25
26 /** 26 /**
27 * 删除队列，根据虚拟主机id 27 * 删除队列，根据虚拟主机id
@@ -112,5 +112,6 @@ public interface BusQueueService { @@ -112,5 +112,6 @@ public interface BusQueueService {
112 * @return 112 * @return
113 */ 113 */
114 List<BusQueue> selectByUsername(String username); 114 List<BusQueue> selectByUsername(String username);
  115 +
115 } 116 }
116 117
@@ -82,7 +82,9 @@ public interface UserInfoService { @@ -82,7 +82,9 @@ public interface UserInfoService {
82 * 修改密码 82 * 修改密码
83 * 83 *
84 * @param userInfo {@link UserInfo} 84 * @param userInfo {@link UserInfo}
85 - * @return 85 + * @return {@link ResultJson}
  86 + * @throws IOException
  87 + * @throws URISyntaxException
86 */ 88 */
87 ResultJson updatePassword(UserInfo userInfo) throws IOException, URISyntaxException; 89 ResultJson updatePassword(UserInfo userInfo) throws IOException, URISyntaxException;
88 90
@@ -102,7 +104,7 @@ public interface UserInfoService { @@ -102,7 +104,7 @@ public interface UserInfoService {
102 * @param userInfo {@link UserInfo} 104 * @param userInfo {@link UserInfo}
103 * @return 返回结果 105 * @return 返回结果
104 */ 106 */
105 - ResultJson deleteUserInfo(UserInfo userInfo) throws IOException, URISyntaxException; 107 + ResultJson deleteUserInfo(UserInfo userInfo) throws IOException, URISyntaxException, TimeoutException, InterruptedException;
106 108
107 /** 109 /**
108 * 删除用户关系 110 * 删除用户关系
1 package com.sunyo.wlpt.message.bus.service.service.impl; 1 package com.sunyo.wlpt.message.bus.service.service.impl;
2 2
3 -import com.sunyo.wlpt.message.bus.service.domain.*;  
4 -import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;  
5 -import com.sunyo.wlpt.message.bus.service.service.*;  
6 -import org.springframework.context.annotation.Lazy; 3 +import com.sunyo.wlpt.message.bus.service.domain.XmlData;
  4 +import com.sunyo.wlpt.message.bus.service.service.MessageNoteService;
7 import org.springframework.scheduling.annotation.Async; 5 import org.springframework.scheduling.annotation.Async;
  6 +import org.springframework.scheduling.annotation.AsyncResult;
8 import org.springframework.stereotype.Service; 7 import org.springframework.stereotype.Service;
9 -import org.springframework.transaction.annotation.Propagation;  
10 -import org.springframework.transaction.annotation.Transactional;  
11 8
12 import javax.annotation.Resource; 9 import javax.annotation.Resource;
13 -import java.io.IOException;  
14 -import java.util.List;  
15 -import java.util.concurrent.TimeoutException; 10 +import java.util.concurrent.CountDownLatch;
  11 +import java.util.concurrent.Future;
16 12
17 /** 13 /**
18 * @author 子诚 14 * @author 子诚
@@ -24,167 +20,24 @@ import java.util.concurrent.TimeoutException; @@ -24,167 +20,24 @@ import java.util.concurrent.TimeoutException;
24 public class AsyncTaskService { 20 public class AsyncTaskService {
25 21
26 @Resource 22 @Resource
27 - private RabbitUtils rabbitUtils;  
28 -  
29 - @Resource  
30 private MessageNoteService messageNoteService; 23 private MessageNoteService messageNoteService;
31 24
32 - @Lazy  
33 - @Resource  
34 - private RoutingKeyService routingKeyService;  
35 -  
36 - @Resource  
37 - private UserMessageBindingService userMessageBindingService;  
38 -  
39 - @Lazy  
40 - @Resource  
41 - private BusQueueService busQueueService;  
42 -  
43 - @Lazy  
44 - @Resource  
45 - private BusExchangeService busExchangeService;  
46 -  
47 - @Lazy  
48 - @Resource  
49 - private VirtualHostService virtualHostService;  
50 -  
51 /** 25 /**
52 * 异步，无论消息是否发送成功，将消息存储于数据库 26 * 异步，无论消息是否发送成功，将消息存储于数据库
53 * 27 *
54 * @param sentData {@link XmlData} 28 * @param sentData {@link XmlData}
55 */ 29 */
56 @Async("taskExecutor") 30 @Async("taskExecutor")
57 - public void saveMessage(XmlData sentData) 31 + public Future<String> saveMessage(XmlData sentData, CountDownLatch latch)
58 { 32 {
  33 + try {
59 // 无论消息是否发送成功，将消息存储于数据库 34 // 无论消息是否发送成功，将消息存储于数据库
60 messageNoteService.insertMessageSelective(sentData); 35 messageNoteService.insertMessageSelective(sentData);
  36 + return new AsyncResult<>("无论消息是否发送成功，将消息存储于数据库与ES中");
  37 + } catch (Exception e) {
  38 + return new AsyncResult<>("保存消息于数据库与ES中，失败");
  39 + } finally {
  40 + latch.countDown();
61 } 41 }
62 -  
63 - /**  
64 - * 当删除服务器的时候,级联删除与服务器有关的  
65 - * 虚拟主机,交换机，路由键，队列，绑定关系  
66 - */  
67 - @Async("taskExecutor")  
68 - @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)  
69 - public void serverCascadeDelete(BusServer busServer) throws IOException, TimeoutException  
70 - {  
71 - String serverId = busServer.getId();  
72 - List<UserMessageBinding> bindings = userMessageBindingService.selectByServerId(serverId);  
73 - // 1.1 删除数据库中的绑定关系  
74 - userMessageBindingService.deleteByServerId(serverId);  
75 - for (UserMessageBinding userMessageBinding : bindings) {  
76 - // 1.2、解除MQ服务器上的绑定关系  
77 - rabbitUtils.toRemoveBinding(userMessageBinding);  
78 - }  
79 - List<VirtualHost> virtualHostList = virtualHostService.selectByServerId(serverId);  
80 - // 2.1、删除数据库中的虚拟主机列表  
81 - virtualHostService.deleteByServerId(serverId);  
82 - for (VirtualHost virtualHost : virtualHostList) {  
83 - String virtualHostId = virtualHost.getId();  
84 - List<BusQueue> queues = busQueueService.selectByVirtualHostId(virtualHostId);  
85 - // 3.1、删除数据库中对应的队列  
86 - busQueueService.deleteByVirtualHostId(virtualHostId);  
87 - for (BusQueue queue : queues) {  
88 - // 3.2、删除MQ服务器上的队列  
89 - rabbitUtils.toRemoveQueue(queue);  
90 - }  
91 - List<BusExchange> exchanges = busExchangeService.selectByVirtualHostId(virtualHostId);  
92 - // 4.1 根据虚拟主机id，删除交换机  
93 - busExchangeService.deleteByVirtualHostId(virtualHostId);  
94 - for (BusExchange busExchange : exchanges) {  
95 - String exchangeId = busExchange.getId();  
96 - // 5.1、删除数据库中对应的路由键  
97 - routingKeyService.deleteByExchangeId(exchangeId);  
98 - // 4.2 在MQ服务器上删除该交换机  
99 - rabbitUtils.toRemoveExchange(busExchange);  
100 - }  
101 - }  
102 - }  
103 -  
104 - /**  
105 - * 当删除虚拟主机的时候，级联删除与虚拟主机有关的  
106 - * 交换机，路由键，队列，绑定关系  
107 - */  
108 - @Async("taskExecutor")  
109 - @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)  
110 - public void virtualHostCascadeDelete(VirtualHost virtualHost) throws IOException, TimeoutException  
111 - {  
112 - String virtualHostId = virtualHost.getId();  
113 - List<UserMessageBinding> bindings = userMessageBindingService.selectByVirtualHostId(virtualHostId);  
114 - // 1.1、删除数据库中的绑定关系  
115 - userMessageBindingService.deleteByVirtualHostId(virtualHostId);  
116 - for (UserMessageBinding userMessageBinding : bindings) {  
117 - // 1.2、解除MQ服务器上的绑定关系  
118 - rabbitUtils.toRemoveBinding(userMessageBinding);  
119 - }  
120 - List<BusQueue> queues = busQueueService.selectByVirtualHostId(virtualHostId);  
121 - // 2.1、删除数据库中对应的队列  
122 - busQueueService.deleteByVirtualHostId(virtualHostId);  
123 - for (BusQueue queue : queues) {  
124 - // 2.2、删除MQ服务器上的队列  
125 - rabbitUtils.toRemoveQueue(queue);  
126 - }  
127 - List<BusExchange> exchanges = busExchangeService.selectByVirtualHostId(virtualHostId);  
128 - // 3.1 根据虚拟主机id，删除交换机  
129 - busExchangeService.deleteByVirtualHostId(virtualHostId);  
130 - for (BusExchange busExchange : exchanges) {  
131 - String exchangeId = busExchange.getId();  
132 - // 4.1、删除数据库中对应的路由键  
133 - routingKeyService.deleteByExchangeId(exchangeId);  
134 - // 3.2 在MQ服务器上删除该交换机  
135 - rabbitUtils.toRemoveExchange(busExchange);  
136 - }  
137 - }  
138 -  
139 - /**  
140 - * 当删除交换机的时候，  
141 - * <p>  
142 - * 同时，删除对应的路由键，删除包含交换机的绑定关系  
143 - * <p>  
144 - * 同时，删除包含该路由键的绑定关系  
145 - */  
146 - @Async("taskExecutor")  
147 - @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)  
148 - public void exchangeCascadeDelete(BusExchange busExchange)  
149 - {  
150 - String exchangeId = busExchange.getId();  
151 - List<RoutingKey> routingKeyList = routingKeyService.selectByExchangeId(exchangeId);  
152 - // 删除相关路由键  
153 - routingKeyService.deleteByExchangeId(exchangeId);  
154 - // 删除交换机相关绑定关系  
155 - userMessageBindingService.deleteByExchangeId(exchangeId);  
156 - // 删除路由键相关绑定关系  
157 - for (int i = 0; i < routingKeyList.size(); i++) {  
158 - userMessageBindingService.deleteByRoutingKeyId(routingKeyList.get(0).getId());  
159 - }  
160 - }  
161 -  
162 - /**  
163 - * 当删除路由键的时候，删除包含该路由键的绑定关系  
164 - */  
165 - @Async("taskExecutor")  
166 - public void routingKeyCascadeDelete(RoutingKey routingKey) throws IOException, TimeoutException  
167 - {  
168 - String routingKeyId = routingKey.getId();  
169 - // 根据路由键id查询出所有的包含该路由键的绑定关系  
170 - List<UserMessageBinding> bindings = userMessageBindingService.selectByRoutingKeyId(routingKeyId);  
171 - for (UserMessageBinding userMessageBinding : bindings) {  
172 - // 解除MQ服务器上的绑定关系  
173 - rabbitUtils.toRemoveBinding(userMessageBinding);  
174 - }  
175 - // 删除数据库中的包含该路由键相关绑定关系  
176 - userMessageBindingService.deleteByRoutingKeyId(routingKey.getId());  
177 } 42 }
178 -  
179 - /**  
180 - * 当删除队列的时候，删除包含队列的绑定关系  
181 - */  
182 - @Async("taskExecutor")  
183 - public void queueCascadeDelete(BusQueue busQueue)  
184 - {  
185 - // 删除相关绑定关系  
186 - userMessageBindingService.deleteByQueueId(busQueue.getId());  
187 - }  
188 -  
189 -  
190 } 43 }
@@ -3,9 +3,13 @@ package com.sunyo.wlpt.message.bus.service.service.impl; @@ -3,9 +3,13 @@ package com.sunyo.wlpt.message.bus.service.service.impl;
3 import com.github.pagehelper.PageHelper; 3 import com.github.pagehelper.PageHelper;
4 import com.github.pagehelper.PageInfo; 4 import com.github.pagehelper.PageInfo;
5 import com.sunyo.wlpt.message.bus.service.domain.BusExchange; 5 import com.sunyo.wlpt.message.bus.service.domain.BusExchange;
  6 +import com.sunyo.wlpt.message.bus.service.domain.RoutingKey;
6 import com.sunyo.wlpt.message.bus.service.mapper.BusExchangeMapper; 7 import com.sunyo.wlpt.message.bus.service.mapper.BusExchangeMapper;
7 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils; 8 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
  9 +import com.sunyo.wlpt.message.bus.service.response.ResultJson;
8 import com.sunyo.wlpt.message.bus.service.service.BusExchangeService; 10 import com.sunyo.wlpt.message.bus.service.service.BusExchangeService;
  11 +import com.sunyo.wlpt.message.bus.service.service.RoutingKeyService;
  12 +import com.sunyo.wlpt.message.bus.service.service.UserMessageBindingService;
9 import org.springframework.context.annotation.Lazy; 13 import org.springframework.context.annotation.Lazy;
10 import org.springframework.stereotype.Service; 14 import org.springframework.stereotype.Service;
11 import org.springframework.transaction.annotation.Propagation; 15 import org.springframework.transaction.annotation.Propagation;
@@ -16,6 +20,8 @@ import java.io.IOException; @@ -16,6 +20,8 @@ import java.io.IOException;
16 import java.util.List; 20 import java.util.List;
17 import java.util.concurrent.TimeoutException; 21 import java.util.concurrent.TimeoutException;
18 22
  23 +import static com.sunyo.wlpt.message.bus.service.common.Constant.RESULT_SUCCESS;
  24 +
19 /** 25 /**
20 * @author 子诚 26 * @author 子诚
21 * Description： 27 * Description：
@@ -32,53 +38,79 @@ public class BusExchangeServiceImpl implements BusExchangeService { @@ -32,53 +38,79 @@ public class BusExchangeServiceImpl implements BusExchangeService {
32 private AsyncTaskService asyncTaskService; 38 private AsyncTaskService asyncTaskService;
33 39
34 @Resource 40 @Resource
  41 + private RoutingKeyService routingKeyService;
  42 +
  43 + @Resource
  44 + private UserMessageBindingService userMessageBindingService;
  45 +
  46 + @Resource
35 RabbitUtils rabbitUtils; 47 RabbitUtils rabbitUtils;
36 48
37 @Override 49 @Override
38 @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED) 50 @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
39 - public int deleteByPrimaryKey(String id) throws IOException, TimeoutException 51 + public ResultJson deleteByPrimaryKey(String id) throws IOException, TimeoutException
40 { 52 {
41 - // 判断删除的个数，需被删除的个数是否一致  
42 int index = 0; 53 int index = 0;
43 String splitItem = ","; 54 String splitItem = ",";
44 - //如果id,传过来多个，以','分割，即批量删除  
45 if (id.contains(splitItem)) { 55 if (id.contains(splitItem)) {
46 String[] split = id.split(splitItem); 56 String[] split = id.split(splitItem);
47 for (int i = 0; i < split.length; i++) { 57 for (int i = 0; i < split.length; i++) {
48 BusExchange busExchange = selectByPrimaryKey(split[i]); 58 BusExchange busExchange = selectByPrimaryKey(split[i]);
49 - // 删除数据库的交换机记录  
50 int num = busExchangeMapper.deleteByPrimaryKey(split[i]); 59 int num = busExchangeMapper.deleteByPrimaryKey(split[i]);
51 - // 删除MQ服务器上的交换机  
52 - deleteExchange(busExchange);  
53 - // 级联删除与交换机相关的路由键，配置关系  
54 - asyncTaskService.exchangeCascadeDelete(busExchange); 60 +
  61 + ResultJson resultJson = deleteExchange(busExchange);
  62 + if (!RESULT_SUCCESS.equals(resultJson.getCode())) {
  63 + return resultJson;
  64 + }
  65 + exchangeCascadeDelete(busExchange);
55 if (num > 0) { 66 if (num > 0) {
56 index = index + num; 67 index = index + num;
57 } 68 }
58 } 69 }
59 - if (index == split.length) {  
60 - return 1;  
61 - } else {  
62 - return 0;  
63 - } 70 + return index == split.length
  71 + ? new ResultJson<>("200", "删除MQ交换机，成功")
  72 + : new ResultJson<>("500", "删除MQ交换机，失败");
64 } else { 73 } else {
65 BusExchange busExchange = selectByPrimaryKey(id); 74 BusExchange busExchange = selectByPrimaryKey(id);
66 - // 删除数据库的交换机记录  
67 int num = busExchangeMapper.deleteByPrimaryKey(id); 75 int num = busExchangeMapper.deleteByPrimaryKey(id);
68 - // 删除MQ服务器上的交换机  
69 - deleteExchange(busExchange);  
70 - // 级联删除与交换机相关的路由键，配置关系  
71 - asyncTaskService.exchangeCascadeDelete(busExchange);  
72 - return num; 76 + ResultJson resultJson = deleteExchange(busExchange);
  77 + if (!RESULT_SUCCESS.equals(resultJson.getCode())) {
  78 + return resultJson;
  79 + }
  80 + exchangeCascadeDelete(busExchange);
  81 + return num > 0
  82 + ? new ResultJson<>("200", "删除MQ交换机，成功")
  83 + : new ResultJson<>("500", "删除MQ交换机，失败");
73 } 84 }
74 } 85 }
75 86
76 /** 87 /**
77 * 删除MQ服务器上的交换机 88 * 删除MQ服务器上的交换机
78 */ 89 */
79 - public void deleteExchange(BusExchange busExchange) throws IOException, TimeoutException 90 + public ResultJson deleteExchange(BusExchange busExchange) throws IOException, TimeoutException
80 { 91 {
81 - rabbitUtils.toRemoveExchange(busExchange); 92 + return rabbitUtils.toRemoveExchange(busExchange);
  93 + }
  94 +
  95 + /**
  96 + * 删除交换机时的级联删除
  97 + * <p>
  98 + * 数据库：删除该交换机的绑定关系
  99 + * <p>
  100 + * 数据库：删除该交换机下的路由键，以及该路由键的绑定关系
  101 + *
  102 + * @param busExchange {@link BusExchange}
  103 + */
  104 + public void exchangeCascadeDelete(BusExchange busExchange)
  105 + {
  106 + String exchangeId = busExchange.getId();
  107 +
  108 + List<RoutingKey> routingKeyList = routingKeyService.selectByExchangeId(exchangeId);
  109 + routingKeyService.deleteByExchangeId(exchangeId);
  110 + userMessageBindingService.deleteByExchangeId(exchangeId);
  111 + for (RoutingKey routingKey : routingKeyList) {
  112 + userMessageBindingService.deleteByRoutingKeyId(routingKey.getId());
  113 + }
82 } 114 }
83 115
84 @Override 116 @Override
@@ -88,10 +120,15 @@ public class BusExchangeServiceImpl implements BusExchangeService { @@ -88,10 +120,15 @@ public class BusExchangeServiceImpl implements BusExchangeService {
88 } 120 }
89 121
90 @Override 122 @Override
91 - public int insertSelective(BusExchange record) throws IOException, TimeoutException 123 + public ResultJson insertSelective(BusExchange record) throws IOException, TimeoutException
92 { 124 {
93 - rabbitUtils.toCreateExchange(record);  
94 - return busExchangeMapper.insertSelective(record); 125 + ResultJson resultJson = rabbitUtils.toCreateExchange(record);
  126 + if (!RESULT_SUCCESS.equals(resultJson.getCode())) {
  127 + return resultJson;
  128 + }
  129 + return busExchangeMapper.insertSelective(record) > 0
  130 + ? new ResultJson<>("200", "新增MQ交换机信息，成功")
  131 + : new ResultJson<>("500", "新增MQ交换机信息，失败");
95 } 132 }
96 133
97 @Override 134 @Override
@@ -5,12 +5,12 @@ import com.github.pagehelper.PageInfo; @@ -5,12 +5,12 @@ import com.github.pagehelper.PageInfo;
5 import com.sunyo.wlpt.message.bus.service.domain.BusQueue; 5 import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
6 import com.sunyo.wlpt.message.bus.service.mapper.BusQueueMapper; 6 import com.sunyo.wlpt.message.bus.service.mapper.BusQueueMapper;
7 import com.sunyo.wlpt.message.bus.service.mapper.UserInfoMapper; 7 import com.sunyo.wlpt.message.bus.service.mapper.UserInfoMapper;
  8 +import com.sunyo.wlpt.message.bus.service.mapper.UserMessageBindingMapper;
8 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils; 9 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
9 import com.sunyo.wlpt.message.bus.service.response.ResultJson; 10 import com.sunyo.wlpt.message.bus.service.response.ResultJson;
10 import com.sunyo.wlpt.message.bus.service.service.BusQueueService; 11 import com.sunyo.wlpt.message.bus.service.service.BusQueueService;
11 import com.sunyo.wlpt.message.bus.service.utils.IdUtils; 12 import com.sunyo.wlpt.message.bus.service.utils.IdUtils;
12 import io.netty.util.internal.StringUtil; 13 import io.netty.util.internal.StringUtil;
13 -import org.springframework.context.annotation.Lazy;  
14 import org.springframework.stereotype.Service; 14 import org.springframework.stereotype.Service;
15 import org.springframework.transaction.annotation.Propagation; 15 import org.springframework.transaction.annotation.Propagation;
16 import org.springframework.transaction.annotation.Transactional; 16 import org.springframework.transaction.annotation.Transactional;
@@ -38,13 +38,14 @@ public class BusQueueServiceImpl implements BusQueueService { @@ -38,13 +38,14 @@ public class BusQueueServiceImpl implements BusQueueService {
38 @Resource 38 @Resource
39 private UserInfoMapper userInfoMapper; 39 private UserInfoMapper userInfoMapper;
40 40
41 - @Lazy 41 +
42 @Resource 42 @Resource
43 - private AsyncTaskService asyncTaskService; 43 + private UserMessageBindingMapper userMessageBindingMapper;
  44 +
44 45
45 @Override 46 @Override
46 @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED) 47 @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
47 - public int deleteByPrimaryKey(String id) throws IOException, TimeoutException 48 + public int deleteByPrimaryKey(String id) throws IOException, TimeoutException, InterruptedException
48 { 49 {
49 // 判断删除的个数，需被删除的个数是否一致 50 // 判断删除的个数，需被删除的个数是否一致
50 int index = 0; 51 int index = 0;
@@ -59,7 +60,7 @@ public class BusQueueServiceImpl implements BusQueueService { @@ -59,7 +60,7 @@ public class BusQueueServiceImpl implements BusQueueService {
59 // 删除MQ服务器上的队列 60 // 删除MQ服务器上的队列
60 deleteQueue(busQueue); 61 deleteQueue(busQueue);
61 // 级联删除数据库中与队列有关的配置关系 62 // 级联删除数据库中与队列有关的配置关系
62 - asyncTaskService.queueCascadeDelete(busQueue); 63 + queueCascadeDelete(busQueue);
63 if (num > 0) { 64 if (num > 0) {
64 index = index + num; 65 index = index + num;
65 } 66 }
@@ -76,7 +77,7 @@ public class BusQueueServiceImpl implements BusQueueService { @@ -76,7 +77,7 @@ public class BusQueueServiceImpl implements BusQueueService {
76 // 删除MQ服务器上的队列 77 // 删除MQ服务器上的队列
77 deleteQueue(busQueue); 78 deleteQueue(busQueue);
78 // 级联删除数据库中与队列有关的配置关系 79 // 级联删除数据库中与队列有关的配置关系
79 - asyncTaskService.queueCascadeDelete(busQueue); 80 + queueCascadeDelete(busQueue);
80 return num; 81 return num;
81 } 82 }
82 } 83 }
@@ -89,6 +90,16 @@ public class BusQueueServiceImpl implements BusQueueService { @@ -89,6 +90,16 @@ public class BusQueueServiceImpl implements BusQueueService {
89 rabbitUtils.toRemoveQueue(busQueue); 90 rabbitUtils.toRemoveQueue(busQueue);
90 } 91 }
91 92
  93 + /**
  94 + * 删除队列时的级联删除
  95 + *
  96 + * @param busQueue {@link BusQueue}
  97 + */
  98 + public void queueCascadeDelete(BusQueue busQueue)
  99 + {
  100 + int num = userMessageBindingMapper.deleteByQueueId(busQueue.getId());
  101 + }
  102 +
92 @Override 103 @Override
93 public int deleteByVirtualHostId(String virtualHostId) 104 public int deleteByVirtualHostId(String virtualHostId)
94 { 105 {
@@ -3,9 +3,11 @@ package com.sunyo.wlpt.message.bus.service.service.impl; @@ -3,9 +3,11 @@ package com.sunyo.wlpt.message.bus.service.service.impl;
3 import com.github.pagehelper.PageHelper; 3 import com.github.pagehelper.PageHelper;
4 import com.github.pagehelper.PageInfo; 4 import com.github.pagehelper.PageInfo;
5 import com.sunyo.wlpt.message.bus.service.domain.RoutingKey; 5 import com.sunyo.wlpt.message.bus.service.domain.RoutingKey;
  6 +import com.sunyo.wlpt.message.bus.service.domain.UserMessageBinding;
6 import com.sunyo.wlpt.message.bus.service.mapper.RoutingKeyMapper; 7 import com.sunyo.wlpt.message.bus.service.mapper.RoutingKeyMapper;
  8 +import com.sunyo.wlpt.message.bus.service.mapper.UserMessageBindingMapper;
  9 +import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
7 import com.sunyo.wlpt.message.bus.service.service.RoutingKeyService; 10 import com.sunyo.wlpt.message.bus.service.service.RoutingKeyService;
8 -import org.springframework.context.annotation.Lazy;  
9 import org.springframework.stereotype.Service; 11 import org.springframework.stereotype.Service;
10 import org.springframework.transaction.annotation.Propagation; 12 import org.springframework.transaction.annotation.Propagation;
11 import org.springframework.transaction.annotation.Transactional; 13 import org.springframework.transaction.annotation.Transactional;
@@ -24,9 +26,11 @@ import java.util.concurrent.TimeoutException; @@ -24,9 +26,11 @@ import java.util.concurrent.TimeoutException;
24 @Service 26 @Service
25 public class RoutingKeyServiceImpl implements RoutingKeyService { 27 public class RoutingKeyServiceImpl implements RoutingKeyService {
26 28
27 - @Lazy  
28 @Resource 29 @Resource
29 - private AsyncTaskService asyncTaskService; 30 + private RabbitUtils rabbitUtils;
  31 +
  32 + @Resource
  33 + private UserMessageBindingMapper userMessageBindingMapper;
30 34
31 @Resource 35 @Resource
32 private RoutingKeyMapper routingKeyMapper; 36 private RoutingKeyMapper routingKeyMapper;
@@ -43,8 +47,8 @@ public class RoutingKeyServiceImpl implements RoutingKeyService { @@ -43,8 +47,8 @@ public class RoutingKeyServiceImpl implements RoutingKeyService {
43 String[] split = id.split(splitItem); 47 String[] split = id.split(splitItem);
44 for (int i = 0; i < split.length; i++) { 48 for (int i = 0; i < split.length; i++) {
45 RoutingKey routingKey = routingKeyMapper.selectByPrimaryKey(split[i]); 49 RoutingKey routingKey = routingKeyMapper.selectByPrimaryKey(split[i]);
46 - // 异步，级联删除包含该路由键的配置关系  
47 - asyncTaskService.routingKeyCascadeDelete(routingKey); 50 + // 级联删除包含该路由键的配置关系
  51 + routingKeyCascadeDelete(routingKey);
48 int num = routingKeyMapper.deleteByPrimaryKey(split[i]); 52 int num = routingKeyMapper.deleteByPrimaryKey(split[i]);
49 if (num > 0) { 53 if (num > 0) {
50 index = index + num; 54 index = index + num;
@@ -57,12 +61,33 @@ public class RoutingKeyServiceImpl implements RoutingKeyService { @@ -57,12 +61,33 @@ public class RoutingKeyServiceImpl implements RoutingKeyService {
57 } 61 }
58 } else { 62 } else {
59 RoutingKey routingKey = routingKeyMapper.selectByPrimaryKey(id); 63 RoutingKey routingKey = routingKeyMapper.selectByPrimaryKey(id);
60 - // 异步，级联删除包含该路由键的配置关系  
61 - asyncTaskService.routingKeyCascadeDelete(routingKey); 64 + // 级联删除包含该路由键的配置关系
  65 + routingKeyCascadeDelete(routingKey);
62 return routingKeyMapper.deleteByPrimaryKey(id); 66 return routingKeyMapper.deleteByPrimaryKey(id);
63 } 67 }
64 } 68 }
65 69
  70 +
  71 + /**
  72 + * 删除路由键时的级联删除
  73 + *
  74 + * @param routingKey {@link RoutingKey}
  75 + * @throws IOException
  76 + * @throws TimeoutException
  77 + */
  78 + public void routingKeyCascadeDelete(RoutingKey routingKey) throws IOException, TimeoutException
  79 + {
  80 + String routingKeyId = routingKey.getId();
  81 + // 根据路由键id查询出所有的包含该路由键的绑定关系
  82 + List<UserMessageBinding> bindings = userMessageBindingMapper.selectByRoutingKeyId(routingKeyId);
  83 + for (UserMessageBinding userMessageBinding : bindings) {
  84 + // 解除MQ服务器上的绑定关系
  85 + rabbitUtils.toRemoveBinding(userMessageBinding);
  86 + }
  87 + // 删除数据库中的包含该路由键相关绑定关系
  88 + userMessageBindingMapper.deleteByRoutingKeyId(routingKey.getId());
  89 + }
  90 +
66 @Override 91 @Override
67 public int deleteByExchangeId(String exchangeId) 92 public int deleteByExchangeId(String exchangeId)
68 { 93 {
@@ -6,6 +6,7 @@ import com.sunyo.wlpt.message.bus.service.domain.BusQueue; @@ -6,6 +6,7 @@ import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
6 import com.sunyo.wlpt.message.bus.service.domain.BusServer; 6 import com.sunyo.wlpt.message.bus.service.domain.BusServer;
7 import com.sunyo.wlpt.message.bus.service.domain.UserInfo; 7 import com.sunyo.wlpt.message.bus.service.domain.UserInfo;
8 import com.sunyo.wlpt.message.bus.service.domain.VirtualHost; 8 import com.sunyo.wlpt.message.bus.service.domain.VirtualHost;
  9 +import com.sunyo.wlpt.message.bus.service.mapper.BusQueueMapper;
9 import com.sunyo.wlpt.message.bus.service.mapper.UserInfoMapper; 10 import com.sunyo.wlpt.message.bus.service.mapper.UserInfoMapper;
10 import com.sunyo.wlpt.message.bus.service.rabbit.utils.ClientUtils; 11 import com.sunyo.wlpt.message.bus.service.rabbit.utils.ClientUtils;
11 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils; 12 import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
@@ -42,6 +43,9 @@ public class UserInfoServiceImpl implements UserInfoService { @@ -42,6 +43,9 @@ public class UserInfoServiceImpl implements UserInfoService {
42 private UserInfoMapper userInfoMapper; 43 private UserInfoMapper userInfoMapper;
43 44
44 @Resource 45 @Resource
  46 + private BusQueueMapper busQueueMapper;
  47 +
  48 + @Resource
45 private RabbitProperties rabbitProperties; 49 private RabbitProperties rabbitProperties;
46 50
47 @Resource 51 @Resource
@@ -253,15 +257,20 @@ public class UserInfoServiceImpl implements UserInfoService { @@ -253,15 +257,20 @@ public class UserInfoServiceImpl implements UserInfoService {
253 @Override 257 @Override
254 public ResultJson updatePassword(UserInfo userInfo) throws IOException, URISyntaxException 258 public ResultJson updatePassword(UserInfo userInfo) throws IOException, URISyntaxException
255 { 259 {
256 - if (StringUtil.isNullOrEmpty(userInfo.getUsername()) || StringUtil.isNullOrEmpty(userInfo.getPassword())) { 260 + String username = userInfo.getUsername();
  261 + String newPassword = userInfo.getPassword();
  262 + if (StringUtil.isNullOrEmpty(username) || StringUtil.isNullOrEmpty(newPassword)) {
257 return new ResultJson<>("400", "用户名和密码，不能为空"); 263 return new ResultJson<>("400", "用户名和密码，不能为空");
258 } 264 }
259 - if (userInfoMapper.selectUserExist(userInfo.getUsername()).size() == 0) { 265 + if (userInfoMapper.selectUserExist(username).size() == 0) {
260 return new ResultJson<>("400", "该用户不存在"); 266 return new ResultJson<>("400", "该用户不存在");
261 } 267 }
262 - BusServer busServer = busServerService.selectByPrimaryKey(userInfo.getServerId());  
263 - String newPassword = userInfo.getPassword(); 268 +
  269 + List<UserInfo> userInfoList = userInfoMapper.selectUserInfoList(userInfo);
  270 + for (UserInfo user : userInfoList) {
  271 + BusServer busServer = busServerService.selectByPrimaryKey(user.getServerId());
264 ClientUtils.updatePassword(busServer, userInfo.getUsername(), newPassword); 272 ClientUtils.updatePassword(busServer, userInfo.getUsername(), newPassword);
  273 + }
265 return new ResultJson<>("200", "修改密码成功"); 274 return new ResultJson<>("200", "修改密码成功");
266 } 275 }
267 276
@@ -277,34 +286,51 @@ public class UserInfoServiceImpl implements UserInfoService { @@ -277,34 +286,51 @@ public class UserInfoServiceImpl implements UserInfoService {
277 } 286 }
278 287
279 /** 288 /**
280 - * 删除根据用户名删除该用户的所有关系 289 + * 删除根据用户名删除该用户的所有关系（只删除当前MQ服务器的该用户）
281 * <p> 290 * <p>
282 * QM服务器上删除该用户（会自动删除该用户关系） 291 * QM服务器上删除该用户（会自动删除该用户关系）
283 * <p> 292 * <p>
284 - * MQ服务器（以及数据库）删除该用户对应队列（日后需要时,打开注释,即可） 293 + * MQ服务器（以及数据库）删除该用户对应队列
285 */ 294 */
286 @Override 295 @Override
287 - public ResultJson deleteUserInfo(UserInfo userInfo) throws IOException, URISyntaxException 296 + @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
  297 + public ResultJson deleteUserInfo(UserInfo userInfo) throws IOException, URISyntaxException, TimeoutException, InterruptedException
288 { 298 {
289 String username = userInfo.getUsername(); 299 String username = userInfo.getUsername();
290 - List<UserInfo> userInfoList = userInfoMapper.selectByUsername(userInfo.getUsername());  
291 - int num = userInfoMapper.deleteByUsername(username); 300 + String serverName = userInfo.getServerName();
  301 + String virtualHostId = userInfo.getVirtualHostId();
  302 + String splitItem = ",";
  303 +
  304 + List<UserInfo> userInfoList = userInfoMapper.selectByUsernameAndServerName(username, serverName);
  305 + int num = userInfoMapper.deleteByUsernameAndServerName(username, serverName);
292 for (UserInfo item : userInfoList) { 306 for (UserInfo item : userInfoList) {
293 BusServer busServer = busServerService.selectByPrimaryKey(item.getServerId()); 307 BusServer busServer = busServerService.selectByPrimaryKey(item.getServerId());
294 ClientUtils.deleteMQUser(username, busServer); 308 ClientUtils.deleteMQUser(username, busServer);
295 } 309 }
296 310
297 -  
298 -// List<BusQueue> queueList = busQueueService.selectByUsername(username);  
299 -// for (BusQueue queue : queueList) {  
300 -// busQueueService.deleteByPrimaryKey(queue.getId());  
301 -// }  
302 - 311 + /**
  312 + * TODO: 删除该用户对应的队列
  313 + */
  314 + if (virtualHostId.contains(splitItem)) {
  315 + String[] splitHostId = virtualHostId.split(splitItem);
  316 + for (int i = 0; i < splitHostId.length; i++) {
  317 + List<BusQueue> queueList = busQueueMapper.selectByUsernameAndHostId(username, splitHostId[i]);
  318 + for (BusQueue queue : queueList) {
  319 + busQueueService.deleteByPrimaryKey(queue.getId());
  320 + }
  321 + }
  322 + } else {
  323 + List<BusQueue> queueList = busQueueMapper.selectByUsernameAndHostId(username, virtualHostId);
  324 + for (BusQueue queue : queueList) {
  325 + busQueueService.deleteByPrimaryKey(queue.getId());
  326 + }
  327 + }
303 return num > 0 328 return num > 0
304 ? new ResultJson<>("200", "删除用户信息,成功!") 329 ? new ResultJson<>("200", "删除用户信息,成功!")
305 : new ResultJson<>("500", "删除用户信息,失败!"); 330 : new ResultJson<>("500", "删除用户信息,失败!");
306 } 331 }
307 332
  333 +
308 /** 334 /**
309 * 删除用户关系（数据库 and MQ服务器） 335 * 删除用户关系（数据库 and MQ服务器）
310 * 336 *
@@ -314,6 +340,7 @@ public class UserInfoServiceImpl implements UserInfoService { @@ -314,6 +340,7 @@ public class UserInfoServiceImpl implements UserInfoService {
314 * @throws URISyntaxException URI语法异常 340 * @throws URISyntaxException URI语法异常
315 */ 341 */
316 @Override 342 @Override
  343 + @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
317 public ResultJson deleteUserRelation(UserInfo userInfo) throws IOException, URISyntaxException 344 public ResultJson deleteUserRelation(UserInfo userInfo) throws IOException, URISyntaxException
318 { 345 {
319 String virtualHostName = userInfo.getVirtualHostName(); 346 String virtualHostName = userInfo.getVirtualHostName();
@@ -326,8 +353,6 @@ public class UserInfoServiceImpl implements UserInfoService { @@ -326,8 +353,6 @@ public class UserInfoServiceImpl implements UserInfoService {
326 353
327 } 354 }
328 } 355 }
329 -  
330 -  
331 int num = userInfoMapper.deleteByPrimaryKey(userInfo.getId()); 356 int num = userInfoMapper.deleteByPrimaryKey(userInfo.getId());
332 BusServer busServer = busServerService.selectByPrimaryKey(userInfo.getServerId()); 357 BusServer busServer = busServerService.selectByPrimaryKey(userInfo.getServerId());
333 ClientUtils.clearPermissions(busServer, userInfo.getVirtualHostName(), userInfo.getUsername()); 358 ClientUtils.clearPermissions(busServer, userInfo.getVirtualHostName(), userInfo.getUsername());
@@ -270,4 +270,14 @@ @@ -270,4 +270,14 @@
270 from bus_queue 270 from bus_queue
271 where username = #{username,jdbcType=VARCHAR} 271 where username = #{username,jdbcType=VARCHAR}
272 </select> 272 </select>
  273 +
  274 + <select id="selectByUsernameAndHostId" parameterType="java.lang.String" resultMap="BaseResultMap">
  275 + select
  276 + <include refid="Base_Column_List"/>
  277 + from bus_queue
  278 + <where>
  279 + username = #{username,jdbcType=VARCHAR}
  280 + and virtual_host_id = #{virtualHostId,jdbcType=VARCHAR}
  281 + </where>
  282 + </select>
273 </mapper> 283 </mapper>
@@ -42,6 +42,15 @@ @@ -42,6 +42,15 @@
42 where username = #{username,jdbcType=VARCHAR} 42 where username = #{username,jdbcType=VARCHAR}
43 </delete> 43 </delete>
44 44
  45 + <delete id="deleteByUsernameAndServerName" parameterType="java.lang.String">
  46 + delete
  47 + from user_info
  48 + <where>
  49 + username = #{username,jdbcType=VARCHAR}
  50 + and server_name= #{serverName,jdbcType=VARCHAR}
  51 + </where>
  52 + </delete>
  53 +
45 <delete id="deleteByServerName" parameterType="java.lang.String"> 54 <delete id="deleteByServerName" parameterType="java.lang.String">
46 delete 55 delete
47 from user_info 56 from user_info
@@ -276,4 +285,14 @@ @@ -276,4 +285,14 @@
276 </if> 285 </if>
277 </where> 286 </where>
278 </select> 287 </select>
  288 +
  289 + <select id="selectByUsernameAndServerName" parameterType="java.lang.String" resultMap="BaseResultMap">
  290 + select
  291 + <include refid="Base_Column_List"/>
  292 + from user_info
  293 + <where>
  294 + username = #{username,jdbcType=VARCHAR}
  295 + and server_name = #{serverName,jdbcType=VARCHAR}
  296 + </where>
  297 + </select>
279 </mapper> 298 </mapper>