优化消息队列

dmp-alarm/pom.xml

@@ -106,6 +106,14 @@
             <artifactId>hystrix-javanica</artifactId>
             <version>1.5.12</version>
         </dependency>
+
+        <!--hutool工具类-->
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+            <version>5.4.3</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

dmp-alarm/src/main/java/com/persagy/dmp/alarm/controller/AlarmCommentController.java

@@ -3,8 +3,10 @@ package com.persagy.dmp.alarm.controller;
 import com.persagy.common.criteria.JacksonCriteria;
 import com.persagy.common.web.MapResponse;
 import com.persagy.common.web.PagedResponse;
+import com.persagy.dmp.alarm.jms.MessageProcesser;
 import com.persagy.dmp.alarm.model.AlarmCommentModel;
 import com.persagy.dmp.alarm.service.AlarmCommentService;
+import com.persagy.dmp.rwd.model.DmpMessage;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -15,6 +17,8 @@ import org.springframework.web.bind.annotation.RestController;
 public class AlarmCommentController {
 	@Autowired
 	private AlarmCommentService service;
+	@Autowired
+	private MessageProcesser messageProcesser;
 	
 	@PostMapping("/query")
 	public PagedResponse<AlarmCommentModel> query(@RequestBody JacksonCriteria criteria) {
@@ -26,10 +30,16 @@ public class AlarmCommentController {
 		return service.create(param);
 	}
 	
-//	@PostMapping("/update")
-//	public MapResponse update(@RequestBody AlarmCommentModel param){
-//		return service.update(param);
-//	}
+	@PostMapping("/sendMessage")
+	public MapResponse update(@RequestBody AlarmCommentModel param){
+		for (int i = 0; i < 2000; i++) {
+			DmpMessage dmpMessage = new DmpMessage();
+			dmpMessage.setMid(String.valueOf(i));
+			dmpMessage.setType("sleep");
+			messageProcesser.convertAndSend(dmpMessage);
+		}
+		return new MapResponse();
+	}
 //
 //	@PostMapping("/delete")
 //	public MapResponse delete(@RequestBody AlarmCommentModel param){

dmp-alarm/src/main/java/com/persagy/dmp/alarm/jms/ConsumeMessageThreadPool.java

@@ -0,0 +1,65 @@
+package com.persagy.dmp.alarm.jms;
+
+import cn.hutool.core.thread.ExecutorBuilder;
+import com.persagy.common.json.JacksonMapper;
+import com.persagy.dmp.rwd.model.DmpMessage;
+import com.rabbitmq.client.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Message;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * 消费消息线程池
+ *
+ * @author lixing
+ * @version V1.0 2021/5/12 2:53 下午
+ **/
+@Component
+@Slf4j
+public class ConsumeMessageThreadPool {
+    private final ExecutorService es;
+
+    @Autowired
+    private MessageProcesser messageProcesser;
+
+    public ConsumeMessageThreadPool() {
+        es = ExecutorBuilder.create()
+                .setCorePoolSize(5)
+                .setMaxPoolSize(10)
+                .setWorkQueue(new LinkedBlockingQueue<>(10))
+                .setHandler(new ThreadPoolExecutor.CallerRunsPolicy())
+                .build();
+    }
+
+    public void consumeMessage(DmpMessage message, Channel channel, Message msg) {
+        es.execute(() -> {
+            try {
+                log.info("============================== Receive:" + JacksonMapper.toSimpleJson(message));
+                messageProcesser.listen(message);
+                // 手动确认消息已消费
+                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+            } catch (Exception e) {
+                Boolean isRedeliveredFail = msg.getMessageProperties().getRedelivered();
+                // true表示消息已经重复处理失败
+                try {
+                    if (isRedeliveredFail) {
+                        // 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列
+                        log.error("重复消费消息失败，message: {}", message);
+                        channel.basicReject(msg.getMessageProperties().getDeliveryTag(), false);
+                    } else {
+                        // 如果是第一次失败则再次放入队列
+                        // requeue为是否重新回到队列，true重新入队
+                        channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
+                    }
+                } catch (Exception exception) {
+                    log.error("重复消费消息失败", exception);
+                }
+            }
+        });
+    }
+}

dmp-alarm/src/main/java/com/persagy/dmp/alarm/jms/JmsConfig.java

@@ -11,8 +11,6 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import java.io.IOException;
-
 /**
  * @author: yaoll
  * @date: 2020-10-22
@@ -21,84 +19,98 @@ import java.io.IOException;
 @Slf4j
 @Configuration
 public class JmsConfig {
-
-
-	@Value("${persagy.dmp.exchange}")
-	private String exchange;
-
-	@Value("${persagy.dmp.alarm.routingKey}")
-	private String dmpAlarmRoutingKey;
-
-	@Value("${alarm.obj.routingKey}")
-	private String alarmObjRoutingKey;
-
-	@Value("${persagy.dmp.rwd.routingKey}")
-	private String rwdRoutingKey;
-
-	@Value("${persagy.dmp.alarm.queue}")
-	private String alarmQueue;
-
-	/**
-	 * 报警对象队列
-	 */
-	private final String alarmObjQueue = "alarm-obj-queue";
-
-	@Autowired
-	private MessageProcesser messageProcesser;
-
-	@Bean
-	public TopicExchange exchange() {
-		return new TopicExchange(exchange);
-	}
-
-	@Bean
-	public Queue alarmQueue() {
-		return new Queue(alarmQueue, true);
-	}
-
-	@Bean
-	public Binding alarmBinding() {
-		return BindingBuilder.bind(alarmQueue()).to(exchange()).with(dmpAlarmRoutingKey);
-	}
-
-	@Bean
-	public Binding rwdBinding() {
-		return BindingBuilder.bind(alarmQueue()).to(exchange()).with(rwdRoutingKey);
-	}
-
-	@Bean
-	public Queue alarmObjQueue() {
-		return new Queue(alarmObjQueue, true);
-	}
-
-	@Bean
-	public Binding alarmObjBinding() {
-		return BindingBuilder.bind(alarmObjQueue()).to(exchange()).with(alarmObjRoutingKey);
-	}
-
-	@RabbitListener(queues = {alarmObjQueue})    //监听器监听指定的Queue
-	public void process(String message, Channel channel, Message msg) {
-		log.info("============================== Receive:" + message);
-		DmpMessage dmpMessage = JacksonMapper.toObject(message, DmpMessage.class);
-		messageProcesser.listen(dmpMessage);
-		// 手动确认消息已消费
-		try {
-			channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
-		} catch (IOException e) {
-			log.error("消息消费反馈失败", e);
-		}
-	}
-
-	@RabbitListener(queues = "${persagy.dmp.alarm.queue}")    //监听器监听指定的Queue
-	public void process(DmpMessage message, Channel channel, Message msg) {
-		log.info("============================== Receive:" + JacksonMapper.toSimpleJson(message));
-		messageProcesser.listen(message);
-		// 手动确认消息已消费
-		try {
-			channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
-		} catch (IOException e) {
-			log.error("消息消费反馈失败", e);
-		}
-	}
-
+    @Autowired
+    ConsumeMessageThreadPool consumeMessageThreadPool;
+
+    @Value("${persagy.dmp.exchange}")
+    private String exchange;
+
+    @Value("${persagy.dmp.alarm.routingKey}")
+    private String dmpAlarmRoutingKey;
+
+    @Value("${alarm.obj.routingKey}")
+    private String alarmObjRoutingKey;
+
+    @Value("${persagy.dmp.rwd.routingKey}")
+    private String rwdRoutingKey;
+
+    @Value("${persagy.dmp.alarm.queue}")
+    private String alarmQueue;
+
+    @Value("${persagy.dmp.alarm.rwd.queue}")
+    private String alarmRwdQueue;
+
+    @Bean
+    public Queue alarmRwdQueue() {
+        return new Queue(alarmRwdQueue, true);
+    }
+
+    @Bean
+    public Binding alarmRwdBinding() {
+        return BindingBuilder.bind(alarmRwdQueue()).to(exchange()).with(rwdRoutingKey);
+    }
+
+    /**
+     * 报警对象队列
+     */
+    private final String alarmObjQueue = "alarm-obj-queue";
+
+    @Autowired
+    private MessageProcesser messageProcesser;
+
+    @Bean
+    public TopicExchange exchange() {
+        return new TopicExchange(exchange);
+    }
+
+    @Bean
+    public Queue alarmQueue() {
+        return new Queue(alarmQueue, true);
+    }
+
+    @Bean
+    public Binding alarmBinding() {
+        return BindingBuilder.bind(alarmQueue()).to(exchange()).with(dmpAlarmRoutingKey);
+    }
+
+    @Bean
+    public Queue alarmObjQueue() {
+        return new Queue(alarmObjQueue, true);
+    }
+
+    @Bean
+    public Binding alarmObjBinding() {
+        return BindingBuilder.bind(alarmObjQueue()).to(exchange()).with(alarmObjRoutingKey);
+    }
+
+    @RabbitListener(queues = {alarmObjQueue})    //监听器监听指定的Queue
+    public void process(String message, Channel channel, Message msg) throws Exception{
+        // 手动确认消息已消费
+        try {
+            log.info("============================== Receive:" + message);
+            DmpMessage dmpMessage = JacksonMapper.toObject(message, DmpMessage.class);
+            if (dmpMessage != null) {
+                messageProcesser.listen(dmpMessage);
+            }
+
+            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+        } catch (Exception e) {
+            Boolean isRedeliveredFail = msg.getMessageProperties().getRedelivered();
+            // true表示消息已经重复处理失败
+            if (isRedeliveredFail) {
+                // 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列
+                log.error("重复消费消息失败，message: {}", message);
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), false);
+            } else {
+                // 如果是第一次失败则再次放入队列
+                // requeue为是否重新回到队列，true重新入队
+                channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
+            }
+        }
+    }
+
+    @RabbitListener(queues = {"${persagy.dmp.alarm.queue}", "${persagy.dmp.alarm.rwd.queue}"})    //监听器监听指定的Queue
+    public void process(DmpMessage message, Channel channel, Message msg) {
+        consumeMessageThreadPool.consumeMessage(message, channel, msg);
+    }
 }