message_bus / message_bus_service · 提交

转到一个项目

作者 王勇
提交 38997e1ee8100b0c66609489c850aea3fc1ca43e 1 个父辈 1ad09235

操作MQ动态创建创建、解除绑定关系

内嵌 并排对比
正在显示 4 个修改的文件 包含 90 行增加 和 18 行删除
... ... @@ -8,6 +8,8 @@ import com.sunyo.wlpt.message.bus.service.service.UserMessageBindingService;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author 子诚
... ... @@ -71,6 +73,7 @@ public class UserMessageBindingController {
*/
@DeleteMapping("/delete")
public ResultJson deleteUserMessageBinding(@RequestBody UserMessageBinding userMessageBinding)
throws IOException, TimeoutException
{
// 执行删除方法
return userMessageBindingService.deleteByPrimaryKey(userMessageBinding.getId()) > 0
... ... @@ -86,6 +89,7 @@ public class UserMessageBindingController {
*/
@GetMapping("/batchRemove")
public ResultJson batchRemoveUserMessageBinding(String ids)
throws IOException, TimeoutException
{
// 执行批量删除
return userMessageBindingService.deleteByPrimaryKey(ids) > 0
... ... @@ -116,6 +120,7 @@ public class UserMessageBindingController {
*/
@PostMapping("/insert")
public ResultJson insertUserMessageBinding(@RequestBody UserMessageBinding userMessageBinding)
throws IOException, TimeoutException
{
// 执行新增
return userMessageBindingService.insertSelective(userMessageBinding) > 0
... ...
package com.sunyo.wlpt.message.bus.service.rabbit.utils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
... ... @@ -147,9 +146,12 @@ public class RabbitUtils {
{
Connection connection = getConnection(serverIp, serverPort, virtualHostName);
Channel channel = connection.createChannel();
AMQP.Exchange.DeclareOk declareOk = channel.exchangeDeclare(busExchange.getExchangeName(), busExchange.getExchangeType(), busExchange.getDurability(),
busExchange.getAutoDelete(), busExchange.getInternal(), null);
log.info("创建交换机的返回值<----->" + declareOk);
channel.exchangeDeclare(busExchange.getExchangeName(),
busExchange.getExchangeType(),
busExchange.getDurability(),
busExchange.getAutoDelete(),
busExchange.getInternal(),
null);
closeConnectionAndChanel(channel, connection);
}
... ... @@ -166,18 +168,26 @@ public class RabbitUtils {
}
/**
* 添加队列（默认设置参数为null）
* 添加队列（默认设置参数为 null）
*/
public void createQueue(String serverIp, Integer serverPort, String virtualHostName, BusQueue busQueue)
throws IOException, TimeoutException
{
Connection connection = getConnection(serverIp, serverPort, virtualHostName);
Channel channel = connection.createChannel();
channel.queueDeclare(busQueue.getQueueName(), busQueue.getDurability(), false, busQueue.getAutoDelete(), null);
channel.queueDeclare(busQueue.getQueueName(),
busQueue.getDurability(),
false,
busQueue.getAutoDelete(),
null);
closeConnectionAndChanel(channel, connection);
}
/**
* 清空队列 channel.queuePurge(queueName);
*/
/**
* 删除队列 channel.queueDelete(queueName);
*/
public void removeQueue(String serverIp, Integer serverPort, String virtualHostName, BusQueue busQueue)
... ... @@ -190,22 +200,33 @@ public class RabbitUtils {
}
/**
* 创建绑定关系
* 创建绑定
*/
public void createBinding(String serverIp, Integer serverPort, String virtualHostName, UserMessageBinding userMessageBinding)
throws IOException, TimeoutException
{
Connection connection = getConnection(serverIp, serverPort, virtualHostName);
Channel channel = connection.createChannel();
channel.queueBind(userMessageBinding.getQueueName(),
userMessageBinding.getExchangeName(),
userMessageBinding.getRoutingKeyName());
closeConnectionAndChanel(channel, connection);
}
/**
* 清空队列 channel.queuePurge(queueName);
*/
/**
* 解除绑定 channel.queueUnbind("queueName", "exchangeName","routingKey");
*/
public void removeBinding(String serverIp, Integer serverPort, String virtualHostName,
UserMessageBinding userMessageBinding)
throws IOException, TimeoutException
{
Connection connection = getConnection(serverIp, serverPort, virtualHostName);
Channel channel = connection.createChannel();
channel.queueUnbind(userMessageBinding.getQueueName(),
userMessageBinding.getExchangeName(),
userMessageBinding.getRoutingKeyName());
closeConnectionAndChanel(channel, connection);
}
/**
* 前往创建交换机的路上
... ... @@ -248,6 +269,28 @@ public class RabbitUtils {
}
/**
* 前往创建绑定的路上
*/
public void toCreateBinding(UserMessageBinding userMessageBinding)
throws IOException, TimeoutException
{
VirtualHost virtualHost = getVirtualHost(userMessageBinding.getVirtualHostId());
BusServer busServer = getBusServer(virtualHost.getServerId());
createBinding(busServer.getServerIp(), busServer.getServerPort(), virtualHost.getVirtualHostName(), userMessageBinding);
}
/**
* 前往解除绑定的路上
*/
public void toRemoveBinding(UserMessageBinding userMessageBinding)
throws IOException, TimeoutException
{
VirtualHost virtualHost = getVirtualHost(userMessageBinding.getVirtualHostId());
BusServer busServer = getBusServer(virtualHost.getServerId());
removeBinding(busServer.getServerIp(), busServer.getServerPort(), virtualHost.getVirtualHostName(), userMessageBinding);
}
/**
* 根据虚拟主机id，获取虚拟主机信息
*
* @param virtualHostId 虚拟主机id
... ...
... ... @@ -4,6 +4,9 @@ import com.github.pagehelper.PageInfo;
import com.sunyo.wlpt.message.bus.service.domain.UserMessageBinding;
import com.sunyo.wlpt.message.bus.service.domain.XmlData;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author 子诚
* Description：
... ... @@ -16,7 +19,7 @@ public interface UserMessageBindingService {
* @param id primaryKey
* @return deleteCount
*/
int deleteByPrimaryKey(String id);
int deleteByPrimaryKey(String id) throws IOException, TimeoutException;
/**
* 新增
... ... @@ -32,7 +35,7 @@ public interface UserMessageBindingService {
* @param record the record
* @return insert count
*/
int insertSelective(UserMessageBinding record);
int insertSelective(UserMessageBinding record) throws IOException, TimeoutException;
/**
* 根据主键查询
... ...
... ... @@ -4,6 +4,7 @@ import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.sunyo.wlpt.message.bus.service.domain.*;
import com.sunyo.wlpt.message.bus.service.mapper.UserMessageBindingMapper;
import com.sunyo.wlpt.message.bus.service.rabbit.utils.RabbitUtils;
import com.sunyo.wlpt.message.bus.service.service.*;
import com.sunyo.wlpt.message.bus.service.utils.IdUtils;
import lombok.extern.slf4j.Slf4j;
... ... @@ -12,7 +13,9 @@ import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeoutException;
import static com.sunyo.wlpt.message.bus.service.common.Constant.EXIST_UMB;
... ... @@ -26,6 +29,9 @@ import static com.sunyo.wlpt.message.bus.service.common.Constant.EXIST_UMB;
public class UserMessageBindingServiceImpl implements UserMessageBindingService {
@Resource
private RabbitUtils rabbitUtils;
@Resource
private UserInfoService userInfoService;
@Resource
... ... @@ -48,7 +54,7 @@ public class UserMessageBindingServiceImpl implements UserMessageBindingService
@Override
@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
public int deleteByPrimaryKey(String id)
public int deleteByPrimaryKey(String id) throws IOException, TimeoutException
{
// 判断删除的个数，需被删除的个数是否一致
int index = 0;
... ... @@ -58,7 +64,9 @@ public class UserMessageBindingServiceImpl implements UserMessageBindingService
try {
String[] split = id.split(splitItem);
for (int i = 0; i < split.length; i++) {
UserMessageBinding userMessageBinding = selectByPrimaryKey(split[i]);
int num = userMessageBindingMapper.deleteByPrimaryKey(split[i]);
deleteBinding(userMessageBinding);
if (num > 0) {
index = index + num;
}
... ... @@ -73,10 +81,21 @@ public class UserMessageBindingServiceImpl implements UserMessageBindingService
return 0;
}
} else {
return userMessageBindingMapper.deleteByPrimaryKey(id);
UserMessageBinding userMessageBinding = selectByPrimaryKey(id);
int num = userMessageBindingMapper.deleteByPrimaryKey(id);
deleteBinding(userMessageBinding);
return num;
}
}
/**
* 解除MQ服务器上的绑定关系
*/
public void deleteBinding(UserMessageBinding userMessageBinding) throws IOException, TimeoutException
{
rabbitUtils.toRemoveBinding(userMessageBinding);
}
@Override
public int insert(UserMessageBinding record)
{
... ... @@ -85,7 +104,7 @@ public class UserMessageBindingServiceImpl implements UserMessageBindingService
@Override
@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
public int insertSelective(UserMessageBinding userMessageBinding)
public int insertSelective(UserMessageBinding userMessageBinding) throws IOException, TimeoutException
{
int index = 0;
String queueId = userMessageBinding.getQueueId();
... ... @@ -382,7 +401,7 @@ public class UserMessageBindingServiceImpl implements UserMessageBindingService
/**
* 接下来的 校验-填充-添加
*/
public int nextValidateAndFill(UserMessageBinding userMessageBinding)
public int nextValidateAndFill(UserMessageBinding userMessageBinding) throws IOException, TimeoutException
{
String validate = validateBinding(userMessageBinding);
if (EXIST_UMB.equals(validate)) {
... ... @@ -390,7 +409,9 @@ public class UserMessageBindingServiceImpl implements UserMessageBindingService
return 0;
} else if (validate == null) {
// 此处添加MQ服务器上的绑定关系
return userMessageBindingMapper.insertSelective(umb_fillName(userMessageBinding));
UserMessageBinding completeBinding = umb_fillName(userMessageBinding);
rabbitUtils.toCreateBinding(completeBinding);
return userMessageBindingMapper.insertSelective(completeBinding);
} else {
return 0;
}
... ...