aidlab/aida_back
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TASK:generate 添加队列消费者，增加并发量

Browse Source
This commit is contained in:
xupei
2024-01-24 14:44:31 +08:00
parent d4568ad740
commit 1d2cb17d54
6 changed files with 73 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/main/java/com/ai/da/common/RabbitMQ/MQConfig.java
View File

@@ -21,10 +21,10 @@ public class MQConfig {
public MQConfig() {
}
@Bean
FanoutExchange fanoutRasaExchange() {
return new FanoutExchange(GENERATE_EXCHANGE_FANOUT);
}
// @Bean
// FanoutExchange fanoutRasaExchange() {
// return new FanoutExchange(GENERATE_EXCHANGE_FANOUT);
// }
/**
* 创建队列,使用工作模式，不用定义交换机
@@ -37,9 +37,9 @@ public class MQConfig {
/**
* 将队列绑定到交换机上【队列订阅交换机】
*/
@Bean
Binding bindingExchangeRasa() {
return BindingBuilder.bind(queueRasa()).to(fanoutRasaExchange());
}
// @Bean
// Binding bindingExchangeRasa() {
// return BindingBuilder.bind(queueRasa()).to(fanoutRasaExchange());
// }
}

65
src/main/java/com/ai/da/common/RabbitMQ/MQConsumer.java
View File

@@ -42,13 +42,13 @@ public class MQConsumer {
@Value("${redis.key.resultMap}")
private String resultMapKey;
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generate(Message msg, Channel channel) {
public void generate(Message msg, Channel channel, String consumerName) {
log.info("============start listening==========");
long start = System.currentTimeMillis();
GenerateThroughImageTextDTO generateThroughImageTextDTO = JSONObject.parseObject(msg.getBody(), GenerateThroughImageTextDTO.class);
String uniqueId = generateThroughImageTextDTO.getUniqueId();
log.info("From " + consumerName + " : " + uniqueId);
try {
// 2、判断当前消息是否在取消列表中
@@ -71,7 +71,7 @@ public class MQConsumer {
// }
// 将消息从redis排队队列中删除,需保证被消费的消息存储到db之后再从redis删除
redisUtil.removeFromZSet(consumptionOrderKey, uniqueId);
if (!Objects.isNull(generateCollectionVO)){
if (!Objects.isNull(generateCollectionVO)) {
HashMap<String, String> generateResult = new HashMap<>();
generateResult.put(uniqueId, JSONObject.toJSONString(generateCollectionVO));
// 将结果存在redis中 ,为空时不要存
@@ -98,7 +98,64 @@ public class MQConsumer {
// 存redis
redisUtil.addToMap(exceptionMapKey, exceptionInfo);
}
long end = System.currentTimeMillis();
log.info(" task_id： " + uniqueId + "----------" + consumerName + " 执行时长：" + (end - start) + "毫秒");
log.info("============end listening==========");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer1(Message msg, Channel channel) {
generate(msg, channel, "consumer 1");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer2(Message msg, Channel channel) {
generate(msg, channel, "consumer 2");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer3(Message msg, Channel channel) {
generate(msg, channel, "consumer 3");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer4(Message msg, Channel channel) {
generate(msg, channel, "consumer 4");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer5(Message msg, Channel channel) {
generate(msg, channel, "consumer 5");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer6(Message msg, Channel channel) {
generate(msg, channel, "consumer 6");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer7(Message msg, Channel channel) {
generate(msg, channel, "consumer 7");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer8(Message msg, Channel channel) {
generate(msg, channel, "consumer 8");
}
@RabbitListener(queues = MQConfig.GENERATE_QUEUE)
@RabbitHandler
public void generateConsumer9(Message msg, Channel channel) {
generate(msg, channel, "consumer 9");
}
}

15
src/main/java/com/ai/da/common/RabbitMQ/MQPublisher.java
View File

@@ -1,9 +1,7 @@
package com.ai.da.common.RabbitMQ;
import com.ai.da.service.RabbitMQService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@@ -16,8 +14,6 @@ public class MQPublisher {
@Resource
private AmqpTemplate amqpTemplate;
@Resource
private RabbitMQService rabbitMQService;
public void sendGenerateMessage(String mm) {
log.info("send message:" + mm);
@@ -25,15 +21,4 @@ public class MQPublisher {
}
public void getMsgCount() {
//// AMQP.Queue.DeclareOk declareOk = rabbitTemplate.execute(channel -> channel.queueDeclarePassive(MQConfig.GENERATE_QUEUE));
//
// QueueInformation queueInfo = rabbitAdmin.getQueueInfo(MQConfig.GENERATE_QUEUE);
//// assert queueInfo != null;
//
//// System.out.println(declareOk.getMessageCount());
// System.out.println(queueInfo.getMessageCount());
//// return declareOk.getMessageCount();
// return queueInfo.getMessageCount();
}
}