Commit 350ad548 authored by yanzg's avatar yanzg

升级新版本

parent 9d6724cd
......@@ -4,6 +4,7 @@ import com.yanzuoguang.dao.DaoConst;
import com.yanzuoguang.util.YzgError;
import com.yanzuoguang.util.helper.StringHelper;
import com.yanzuoguang.util.helper.YzgTimeout;
import com.yanzuoguang.util.vo.Ref;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
......@@ -82,13 +83,21 @@ public class BeanDao {
* @return 创建成功的队列
*/
public Queue createQueue(String queueName, long deadTime, String deadExchange, String deadRouteKey) {
Ref<Queue> ret = new Ref<>(null);
YzgTimeout.timeOut(BeanDao.class, "创建队列" + queueName, () ->
ret.value = createQueueRun(queueName, deadTime, deadExchange, deadRouteKey)
);
// 重新获取实体
return ret.value;
}
private Queue createQueueRun(String queueName, long deadTime, String deadExchange, String deadRouteKey) {
String key = StringHelper.getId(QUEUE, queueName);
// 判断队列是否存在,不存在则锁定再次判断
Queue queueHis = getBean(Queue.class, key);
if (queueHis != null) {
return queueHis;
}
// 创建队列实体
Map<String, Object> params = new HashMap<>(DaoConst.COLLECTION_INIT_SIZE);
if (!StringHelper.isEmpty(deadExchange) && !StringHelper.isEmpty(deadRouteKey)) {
......@@ -100,14 +109,12 @@ public class BeanDao {
// x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
params.put("x-dead-letter-routing-key", deadRouteKey);
}
// 创建实体
Queue queueNew = new Queue(queueName, true, false, false, params);
YzgTimeout.timeOut(BeanDao.class, "创建队列" + queueName, () -> {
amqpAdmin.declareQueue(queueNew);
});
// 将实体注册到上下文中
register(key, queueNew);
// 重新获取实体
return queueNew;
}
......
......@@ -65,6 +65,7 @@ public class MessageDaoImpl extends BaseDaoImpl implements MessageDao, Initializ
*/
@Override
public void afterPropertiesSet() throws Exception {
ThreadHelper.runThread(() ->
YzgTimeout.timeOut(MessageDaoImpl.class, "消息队列处理工具类初始化", () -> {
List<MapRow> tables = this.getDb().query(MessageDaoImpl.class, "QUERY_TABLE_SQL", QUERY_TABLE_SQL);
if (tables.isEmpty()) {
......@@ -72,7 +73,8 @@ public class MessageDaoImpl extends BaseDaoImpl implements MessageDao, Initializ
} else {
this.getDb().update(MessageDaoImpl.class, "ALTER_TABLE_SQL", ALTER_TABLE_SQL);
}
});
})
);
}
/**
......
......@@ -53,10 +53,14 @@ public class MessageLogDaoImpl extends BaseDaoImpl implements MessageLogDao, Ini
*/
@Override
public void afterPropertiesSet() throws Exception {
ThreadHelper.runThread(() -> {
YzgTimeout.timeOut(MessageLogDaoImpl.class, "消息队列处理工具类初始化", () -> {
List<MapRow> tables = this.getDb().query(MessageLogDaoImpl.class, "QUERY_TABLE_SQL", QUERY_TABLE_SQL);
if (tables.isEmpty()) {
this.getDb().update(MessageLogDaoImpl.class, "CREATE_TABLE_SQL", CREATE_TABLE_SQL);
}
});
});
}
/**
......
......@@ -45,10 +45,14 @@ public class QueueServerDaoImpl extends BaseDaoImpl implements QueueServerDao, I
*/
@Override
public void afterPropertiesSet() throws Exception {
ThreadHelper.runThread(() -> {
YzgTimeout.timeOut(QueueServerDaoImpl.class, "消息队列处理工具类初始化", () -> {
DbExecute db = this.getDb();
List<MapRow> tables = db.query(QueueServerDaoImpl.class, "QUERY_TABLE_SQL", QUERY_TABLE_SQL);
if (tables.isEmpty()) {
db.update(QueueServerDaoImpl.class, "CREATE_TABLE_SQL", CREATE_TABLE_SQL);
}
});
});
}
}
......@@ -48,9 +48,13 @@ public class QueueServerTokenDaoImpl extends BaseDaoImpl implements QueueServerT
*/
@Override
public void afterPropertiesSet() throws Exception {
ThreadHelper.runThread(() -> {
YzgTimeout.timeOut(QueueServerTokenDaoImpl.class, "消息队列处理工具类初始化", () -> {
List<MapRow> tables = this.getDb().query(QueueServerTokenDaoImpl.class, "QUERY_TABLE_SQL", QUERY_TABLE_SQL);
if (tables.isEmpty()) {
this.getDb().update(QueueServerTokenDaoImpl.class, "CREATE_TABLE_SQL", CREATE_TABLE_SQL);
}
});
});
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment